Sping Boot 和@ Kafkafka

gupuwyp2  于 2023-10-15  发布在  Apache
关注(0)|答案(1)|浏览(113)

我使用Sping Boot 2.7与Kafka集成来使用记录。
这是我的听众,很简单:

@KafkaListener(topics = {"journal-topic1"})
public void onMessage(List<ConsumerRecord<String, String>> records) {
    //processing
}

我的问题是:我没有得到List<ConsumerRecord<String, String>>,而是得到了我的值列表:List<String>
下面是我的配置:

spring:
  kafka:
    topic: "journal-topic1"
    properties:
      auto-create-topics-enable: true
    consumer:
      bootstrap-servers: localhost:9092
      group-id: "journal-group1"
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: latest
    admin:
      properties:
        bootstrap-servers: localhost:9092
xuo3flqw

xuo3flqw1#

请给予更多细节,但现在尝试更改您的代码以使用ConsumerRecord而不是List。

@KafkaListener(topics = {"journal-topic1"})
public void message(ConsumerRecord<String, String> record) {
    //Process
}

或者如果你想消费整批消费者记录,在那里查看:
Kafka listener receiving List<ConsumerRecord<String, String>>, is it possible to consume?

相关问题