stream-kafka批量消费

Question

springcloud-stream-kafka 使用批量消费消息的时候（已添加bathc-mode = true），怎么获取到批量消息。如下代码，在获取到消息的时候，老是提示 tokenJosn 不能转String。。有没有兄弟知道怎么解决

@Bean
  public Consumer<Message<List<String>>> myEvent(){
      return message -> {
            List<String> mess= message.getPayload();
      }
  }

Answer 1

Spring Cloud Stream Kafka默认对消息进行字节序列化和反序列化，它将消息载荷作为字节数组处理。当你试图将消息载荷转换为String列表时，可能会出现问题，因为实际上它们是字节数组。

为了解决这个问题，你可以在消费者端配置Spring Cloud Stream Kafka使用String的反序列化器。在你的配置文件（如application.properties或application.yml）中添加以下配置：

spring.cloud.stream.kafka.bindings.myEvent-in-0.consumer.configuration.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

这将配置Kafka消费者使用StringDeserializer反序列化消息的值。

然后，你可以在消费者方法中将消息载荷作为字符串列表处理：

@Bean
public Consumer<List<String>> myEvent() {
    return list -> {
        for (String s : list) {
            // 处理字符串s
        }
    };
}

注意，这里假设myEvent-in-0是你的输入绑定的名称，你需要将其替换为实际的输入绑定名称。同样，如果你的方法名或者函数的签名与上述代码不同，你也需要相应地修改你的配置和代码。