我正在使用Spring Kafka 2.2.7,我已经配置了
我想添加一个
自版本2.2.7开始,可以向监听器容器中添加RecordInterceptor;在调用侦听器之前将调用它,以允许检查或修改记录。
我已经查阅了Spring文档,但没有找到解决方案,这似乎是一件简单的事情,但也许我错过了什么。
如有任何帮助将不胜感激。
提前致谢。
@EnableKafka
和kafkaListenerContainerFactory
,并使用@KafkaListener
来消费消息,一切都按预期运行。我想添加一个
RecordInterceptor
来记录所有被消费的消息,但是我发现很难配置它。 文档指出可以在容器上设置RecordInterceptor,但我不知道如何获取容器的实例。自版本2.2.7开始,可以向监听器容器中添加RecordInterceptor;在调用侦听器之前将调用它,以允许检查或修改记录。
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Bytes> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, Bytes> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(createConsumerFactory());
factory.setConcurrency(consumerCount);
return factory;
}
我已经查阅了Spring文档,但没有找到解决方案,这似乎是一件简单的事情,但也许我错过了什么。
如有任何帮助将不胜感激。
提前致谢。