spring - Kafka 监听器接收列表>，可以食用吗？

Question

我是 Kafka 的超级新手，坦率地说，我对这种类型的消费者一无所知（据我所知，due 是批处理准备好的），所以我正在努力弄清楚如何基本上使用这些事件的列表。

我有这样的事情：

    @KafkaListener(topics = "#{'${kafka.listener.list-of-topics}'.split(',')}")
        public void readMessage(List<ConsumerRecord<String, String>> records,
                                   final Acknowledgment acknowledgment) {

            try {
....

我知道当我收到一个事件（至少一个事件）是“MyObject”类型的，所以当我收到一条消息时我可以做得很好。

我相信必须有一种方法来阅读/投射这个List<ConsumerRecords<String,String>，但我不知道如何..

有任何想法吗？

标签： springkafka-consumer-apispring-kafka

Answer 1

请参阅参考手册：批处理侦听器。

从 1.1 版本开始，@KafkaListener 方法可以配置为接收从消费者轮询接收的整批消费者记录。要配置侦听器容器工厂以创建批处理侦听器，请设置 batchListener 属性：

@Bean
public KafkaListenerContainerFactory<?> batchFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);  // <<<<<<<<<<<<<<<<<<<<<<<<<
    return factory;
}

...

您还可以接收ConsumerRecord<?, ?>对象列表，但它必须是方法上定义的唯一参数（除了 optional Acknowledgment，在使用手动提交时和/或Consumer<?, ?>参数）：

...

使用 Spring Boot 时，设置属性spring.kafka.listener.type=batch。