- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我在尝试使用 Spring Kafka 将 Kafka 消息反序列化到 POJO 时遇到问题。我想使用消息的键和值部分来构造 POJO。
Kafka消息键是一个字符串,消息值是JSON。
我尝试按照 codenotfound.com 中的教程仅执行消息的值部分和 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 应用程序都无法将消息转换为 Data 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
我必须编写自定义反序列化器吗?
自定义解串器会是什么样子?
最佳答案
欢迎来到 StackOverflow!
默认情况下,Spring Kafka 在使用消息时使用字符串反序列化器,因此在您的情况下,您似乎想要反序列化 Json 消息,为此,第一步是将值反序列化器注册为 JsonDeserializer.class。这应该适用于消息的值,但仍然无法解决您也想要的 key 。在 Kafka 中,键和值序列化器没有组合在一起,所以我认为没有一种简单的方法可以在反序列化时获取 key ,您最简单的选择可能是:
创建 Json 对象的关键部分,以便使用 JsonDeserliazer 自动反序列化。
在消费者端进行处理,而不是接收对象本身,而是使用 ConsumerRecord
,它将返回反序列化的键和值,因此您可以简单地使用 setter 将键添加到反序列化的对象中。
我希望它有助于澄清。我将快速浏览一下你在 Github 上的示例并做一个 PR,完成。因此,要使用将 key 作为消息有效负载的一部分的方法来修复它(检查存储库中的 PR):
将 key 作为属性添加到数据对象并供您的使用者使用:
@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,我建议您使用带有单个 Zookeeper 和 kafka 的 docker-compose 文件,请检查此示例 -> https://dev.to/thegroo/spring-kafka-producer-and-consumer-41oc或另一个 -> https://dev.to/thegroo/one-to-run-them-all-1mg6
关于java - 如何将 kafka 消息反序列化到 POJO?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/60553992/
我是一名优秀的程序员,十分优秀!