如何阅读Kafka消息键从Spring的云流？

Question

public class GreetingsListener {

    @StreamListener(GreetingsProcessor.INPUT)
    public void handleGreetings(@Payload Greetings request) {
     
    }
}

Answer 1

您可以尝试这样的方法：

@StreamListener(GreetingsProcessor.INPUT)
public void handleGreetings(@Payload Greetings request, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY)String  key) {

}

您需要为密钥提供适当的反序列化器。例如，如果您的密钥是字符串，那么您可以提供：

spring.cloud.stream.kafka.binder.configuration.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer

如果需要对不同的输入通道使用不同的密钥反序列化器，可以在每个kafka绑定的生产者部分中扩展此设置。例如：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          <channel_name>:
            consumer:
              startOffset: latest
              autoCommitOffset: true
              autoCommitOnError: true
              configuration:
                key.deserializer: org.apache.kafka.common.serialization.StringDeserializer