指定 ListenerContainerFactory
默认情况下,RetryTopic 配置将使用@KafkaListener
注释,但您可以指定一个不同的 Bean Bean 来用于创建 Retry Topic 和 DLT Listener 容器。
对于@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 开始,您可以对可重试和不可重试的主题使用相同的工厂。 |