过滤消息

Doubao 中英对照 Filtering Messages

在某些场景下，例如再均衡，一条已经处理过的消息可能会被重新投递。框架无法知道这样的消息是否已被处理过。那是应用层的功能。这被称为幂等接收器模式，Spring Integration 提供了它的一个实现。

Spring for Apache Kafka 项目还通过 FilteringMessageListenerAdapter 类提供了一些辅助功能，它可以包装你的 MessageListener。该类接收一个 RecordFilterStrategy 的实现，你在其中实现 filter 方法以表明某条消息是重复的、应当被丢弃。它还有一个名为 ackDiscarded 的附加属性，用于指示适配器是否应当确认被丢弃的记录。默认为 false。

当你使用 @KafkaListener 时，在容器工厂上设置 RecordFilterStrategy（以及可选的 ackDiscarded），以便监听器被包装在合适的过滤适配器中。

此外，还提供了一个 FilteringBatchMessageListenerAdapter，用于当你使用批消息监听器时。

important

如果你的 @KafkaListener 接收的是 ConsumerRecords<?, ?> 而不是 List<ConsumerRecord<?, ?>>，则 FilteringBatchMessageListenerAdapter 会被忽略，因为 ConsumerRecords 是不可变的。

从版本 2.8.4 开始，你可以使用监听器注解上的 filter 属性来覆盖监听器容器工厂的默认 RecordFilterStrategy。

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

从版本 3.3 开始，支持忽略因 RecordFilterStrategy 过滤而产生的空批次。在实现 RecordFilterStrategy 时，可以通过 ignoreEmptyBatch() 对其进行配置。默认设置为 false，表示即使所有 ConsumerRecord 都被过滤掉，仍会调用 KafkaListener。

如果返回 true，则当所有 ConsumerRecord 都被过滤掉时，将不会调用 KafkaListener。但是，向 broker 的提交仍会被执行。

如果返回 false，则当所有 ConsumerRecord 都被过滤掉时，仍会调用 KafkaListener。

下面是一些示例。

public class IgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy {
    ...
    @Override
    public List<ConsumerRecord<String, String>> filterBatch(
            List<ConsumerRecord<String, String>> consumerRecords) {
        return List.of();
    }

    @Override
    public boolean ignoreEmptyBatch() {
        return true;
    }
}

// NOTE: ignoreEmptyBatchRecordFilterStrategy is bean name of IgnoreEmptyBatchRecordFilterStrategy instance.
@KafkaListener(id = "filtered", topics = "topic", filter = "ignoreEmptyBatchRecordFilterStrategy")
public void listen(List<Thing> things) {
    ...
}

在这种情况下，IgnoreEmptyBatchRecordFilterStrategy 始终返回空列表，并返回 true 作为 ignoreEmptyBatch() 的结果。因此 KafkaListener#listen(…​) 根本不会被调用。

public class NotIgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy {
    ...
    @Override
    public List<ConsumerRecord<String, String>> filterBatch(
            List<ConsumerRecord<String, String>> consumerRecords) {
        return List.of();
    }

    @Override
    public boolean ignoreEmptyBatch() {
        return false;
    }
}

// NOTE: notIgnoreEmptyBatchRecordFilterStrategy is bean name of NotIgnoreEmptyBatchRecordFilterStrategy instance.
@KafkaListener(id = "filtered", topics = "topic", filter = "notIgnoreEmptyBatchRecordFilterStrategy")
public void listen(List<Thing> things) {
    ...
}

然而，在这种情况下，NotIgnoreEmptyBatchRecordFilterStrategy 始终返回空列表，并返回 false 作为 ignoreEmptyBatch() 的结果。因此 KafkaListener#listen(…​) 始终会被调用。