Spring Cloud Stream中json字符串到对象的自动转换是如何工作的？答案

【问题标题】：How does json String to Object auto conversion works in Spring cloud Stream?Spring Cloud Stream中json字符串到对象的自动转换是如何工作的？
【发布时间】：2018-03-29 15:28:54
【问题描述】：

我在看这个例子 - https://github.com/spring-cloud/spring-cloud-stream-samples/blob/master/kafka-streams-samples/kafka-streams-product-tracker/src/main/java/kafka/streams/product/tracker/KafkaStreamsProductTrackerApplication.java

尝试做类似的事情，但对我来说它不起作用。产品 json 字符串如何作为产品对象接收？

【问题讨论】：

标签： apache-kafka-streams spring-cloud-stream

【解决方案1】：

默认情况下，入站 KStream 的反序列化由 Spring Cloud Stream 完成。使用的默认内容类型是application/json（相当于提供属性：spring.cloud.stream.bindings.input.contentType: application/json）。这就是正确转换产品 json 字符串的原因。

您可以禁用框架级别的转换并让 Kafka 执行此操作，在这种情况下您需要通过属性提供 Serdes。为了启用本机反序列化，您可以设置属性 - spring.cloud.stream.bindings.input.consumer.useNativeDecoding: true。然后您需要提供适当的 Serdes。有关所有这些的更多信息都在这里：https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.BUILD-SNAPSHOT/reference/htmlsingle/#_message_conversion

【讨论】：

本文档有一个示例绑定为 - spring.cloud.stream.kstream.bindings.output3.producer.valueSerde=JsonSerde。应该是 - spring.cloud.stream.kafka.streams.bindings.output3.producer.valueSerde=JsonSerde
你是对的。应该是spring.cloud.stream.kafka.streams.bindings.....。感谢您指出了这一点。将在文档中修复它。
我创建了自定义 Serde - CustomSerde 扩展了 JsonSerde 出于某种原因 - 它无法加载 valueSerde 进行绑定，但它尝试仅加载默认值 serde。这失败了 - 无法将字符串转换为自定义对象。 spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde：Serdes$StringSerde spring.cloud.stream.kafka.streams.bindings.mysql-kstream-member.consumer.valueSerde：CustomSerde 这有效 - 春天.cloud.stream.kafka.streams.binder.configuration.default.value.serde: CustomSerde
似乎 valueSerde 适用于 spring-cloud-stream 应用程序中发送的任何消息，但不适用于从外部应用程序接收的消息。即使有 Kafka-streams 绑定，它也会忽略它并寻找 binder 默认值 serve。
我得到了这个工作，看起来 AvroMapper 正在做一些问题。我删除了映射逻辑以使用 AvroMapper 并且它有效。不知道是什么。