指定 ListenerContainerFactory
預設情況下,RetryTopic 配置將使用 @KafkaListener
註解中提供的工廠,但您可以指定一個不同的工廠來建立重試主題和 DLT 監聽器容器。
對於 @RetryableTopic
註解,您可以提供工廠的 bean 名稱,並且使用 RetryTopicConfiguration
bean,您可以提供 bean 名稱或例項本身。
@RetryableTopic(listenerContainerFactory = "my-retry-topic-factory")
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
// ... message processing
}
@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template,
ConcurrentKafkaListenerContainerFactory<Integer, MyPojo> factory) {
return RetryTopicConfigurationBuilder
.newInstance()
.listenerFactory(factory)
.create(template);
}
@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
return RetryTopicConfigurationBuilder
.newInstance()
.listenerFactory("my-retry-topic-factory")
.create(template);
}
自 2.8.3 起,您可以為可重試和不可重試主題使用同一個工廠。 |