如何将 kafka 消息反序列化为 POJO？

Question

我在尝试使用 Spring Kafka 将 Kafka 消息反序列化为 POJO 时遇到问题。我想使用消息的键和值部分来构造POJO。

Kafka消息键为字符串，消息值为JSON。

我尝试按照 codenotfound.com and baeldung.com 上的教程只处理消息的值部分。除了我还想在 POJO 中拥有键值并且 java 应用程序没有生成消息。

如何让 java 应用程序将 kafka 消息适当地反序列化为 POJO？

例如：

key = "test"

{
  "value1": "1st value"
  "value2": "2nd value"
}

可以在以下位置找到我正在尝试的可重现示例： https://github.com/gl3h/Simple-Consumer

要重现该问题，必须执行以下操作：

运行命令 docker-compose up -d 调出 3 个 Zookeeper 和 Kafka 实例。它还会启动连接到 Kafka 集群的 Kafdrop。
运行 java 申请。 (gradle bootrun)

向数据主题发送消息

kafka-console-producer --broker-list kafka1:29092 --topic data --property "parse.key=true" --property "key.separator=&" 
test&{"value1":"1st value","value2":"2nd value"}```

每当向 Kafka 集群发送消息时，Java 应用程序无法将消息转换为数据 POJO，并出现以下错误：

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.example.consumer.example.ExampleConsumer.processData(com.example.consumer.example.Data)]
Bean [com.example.consumer.example.ExampleConsumer@7a5a16cf]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}], failedMessage=GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}], failedMessage=GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1641) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1630) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1546) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1487) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1401) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1165) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:949) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:884) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_221]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_221]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_221]
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}], failedMessage=GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:314) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:86) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1592) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1575) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1534) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    ... 8 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]
    at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor$KafkaNullAwarePayloadArgumentResolver.resolveArgument(KafkaListenerAnnotationBeanPostProcessor.java:905) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:117) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:148) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:48) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:304) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    ... 13 common frames omitted

我是否必须编写自定义解串器？
自定义反序列化器会是什么样子？

Answer 1

欢迎使用 Whosebug！

默认情况下 Spring Kafka 在使用消息时使用字符串反序列化器，因此在您的情况下，您似乎想要反序列化 Json 消息，为此第一步是注册作为值反序列化器为 JsonDeserializer.class。这应该适用于消息的值，但仍然不能解决您也想要的密钥。

在 Kafka 中，Key 和 Value 序列化器没有组合在一起，所以我认为没有一种简单的方法可以在反序列化时同时获取密钥，最简单的选择可能是：

创建 Json 对象的关键部分，以便使用 JsonDeserliazer 自动反序列化。
消费者端的进程接收而不是对象本身，而是使用 ConsumerRecord 这将 return 键和值反序列化，因此您可以简单地将键添加到使用 setter 反序列化对象。

我希望它有助于澄清。我将快速查看您在 Github 上的示例并进行 PR，完成。因此，要使用将密钥作为消息有效负载的一部分的方法来修复它（检查您的 Repo 中的 PR）：

将密钥作为属性添加到数据对象，对于您的消费者：

 @Component
public class ExampleConsumer {

    @KafkaListener(topics = "data")
    public void processData(Data data) {
        System.out.println("Data:" + data);
    }
}

并添加正确的配置，application.yml:

spring:
  application:
    name: kafka-consumer-example
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      group-id: data_consumer
      client-id: ${spring.application.name}
      properties:
        spring.json.value.default.type: com.example.consumer.example.Data
        spring.json.type.mapping: "data:com.example.consumer.example.Data"
        spring.json.trusted.packages: "*"

    listener:
      missing-topics-fatal: false

P.S - 对于本地的运行 kafka，我建议你使用一个 docker-compose 文件和一个 zookeeper 和 kafka，检查这个例子 -> https://dev.to/thegroo/spring-kafka-producer-and-consumer-41oc or this other one -> https://dev.to/thegroo/one-to-run-them-all-1mg6

Answer 2

您的侦听器未使用您的自定义容器工厂；将其命名为 kafkaListenerContainerFactory 以替换引导程序或在侦听器上使用 containerFactory 属性。

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Data> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Data> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(dataConsumerFactory());
    return factory;
}

和

Data:Data{key='null', value1='1st value', value2='2nd value'}

或者，更简单，通过属性配置启动。

如何将 kafka 消息反序列化为 POJO？

How do I deserialize a kafka message to a POJO?

java

apache-kafka

spring-kafka

spring-messaging