使用过滤策略和手动提交的spring-kafka批处理示例

Question

我计划使用SpringKafka批处理侦听器进行批处理。我正在寻找这两种方案的几个示例。

1. 我们如何通过批处理实现过滤记录策略？更新 ：来自文档 - “此外，还提供了过滤批处理消息警报器适配器，用于使用批处理消息侦听器时。我没有看到任何容器工厂方法来设置此筛选器批处理消息驱动程序对象或筛选器实现。

以下是我的批处理侦听器过滤策略代码:

@Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory) {

        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<Object, Object>();
        configurer.configure(factory, kafkaConsumerFactory);
        factory.setBatchListener(true);
        factory.setAckDiscarded(true);
        factory.setRecordFilterStrategy(new RecordFilterStrategy<Object, Object>() {
            @Override
            public boolean filter(ConsumerRecord<Object, Object> consumerRecords) {

                //log.info("Retrieved the record {} from the partition {} with offset {}", consumerRecord.value(), consumerRecord.partition(), consumerRecord.offset());

                return true;
            }
        });
        
        return factory;
    }

现在我想到的另一个问题是，上述方案如何与单个消费者和多个消费者一起工作。

假设案例1:单一消费者

假设我们有一个具有5个分区的主题。当我们订阅该主题时，我们假设我们从该主题中获得了100条消息，其中每个分区有20条消息。如果我们想提交这些消息偏移量，确认对象是否保存了最后一条消息的每个分区和最后一个偏移量？

案例2：多个消费者

对于案例1中提到的相同输入，如果我们启用分区计数的相等的消费者数量，ack对象是否保存分区和最后一条消息的偏移量？

你能帮我解决这个问题吗？

Answer 1

>

请参见过滤BatchMessageListenerAdapterhttps://docs.spring.io/spring-kafka/docs/current/reference/html/#filtering-消息

处理批处理异常的最简单方法是将RecoveringBatchErrorHandler与DeadLetterPublishingRecoverer一起使用。抛出一个＜code＞BatchListenerFailedException＜/code＞以指示批处理中的哪个记录失败；成功记录的偏移量将被提交，剩余记录（包括失败的记录）将被重新提交，直到重试次数（如果配置）耗尽，此时失败的记录将转到死信主题，其余记录将被重新发送。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#recovering-批量eh

是的，当批处理被确认时，将提交批处理中每个分区的最新偏移量（1）。

如果您有多个消费者，分区将分布在这些消费者之间。