对于最新的稳定版本,请使用 Spring for Apache Kafka 3.3.9spring-doc.cadn.net.cn

过滤消息

在某些情况下,例如重新平衡,可能会重新传递已处理的消息。 框架无法知道此类消息是否已被处理。 这是一个应用程序级功能。 这被称为幂等接收器模式,Spring Integration 提供了它的实现spring-doc.cadn.net.cn

Spring for Apache Kafka 项目还通过FilteringMessageListenerAdapter类,它可以将MessageListener. 此类采用RecordFilterStrategy在其中实现filter方法来表示消息是重复的,应该丢弃。 这有一个名为ackDiscarded,指示适配器是否应确认已丢弃的记录。 是的false默认情况下。spring-doc.cadn.net.cn

当您使用@KafkaListener,将RecordFilterStrategy(并且可选ackDiscarded) ,以便侦听器被包装在适当的过滤适配器中。spring-doc.cadn.net.cn

此外,一个FilteringBatchMessageListenerAdapter,用于使用批处理消息侦听器时。spring-doc.cadn.net.cn

FilteringBatchMessageListenerAdapter如果您的@KafkaListener收到一个ConsumerRecords<?, ?>而不是List<ConsumerRecord<?, ?>>因为ConsumerRecords是不可变的。

从 2.8.4 版本开始,您可以覆盖侦听器容器工厂的默认值RecordFilterStrategy通过使用filter属性。spring-doc.cadn.net.cn

@KafkaListener(id = "filtered", topics = "topic", filter = "differentFilter")
public void listen(Thing thing) {
    ...
}