如何处理 Kafka 主题中的 JSON？答案

【问题标题】：How to handle JSON in a Kafka topic?如何处理 Kafka 主题中的 JSON？
【发布时间】：2018-12-18 00:04:26
【问题描述】：

我正在构建一个 Java 应用程序，其中有一些 JSON 对象（特别是来自 com.satori.rtm.model.AnyJson 的 AnyJson 对象）并且我想要在 Kafka 主题中发送这些对象。我应该以 AnyJson o String 类型发送它们吗？我问这个是因为构造函数KafkaProducer<K, V> 在（反）序列化时处理 JSON 值似乎有一些问题。

这在我的生产者配置中

Properties props= new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokerEndpoint);
props.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducer");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
kafkaProducer= new KafkaProducer<Integer, AnyJson>(props);

其中 JsonSerializer.class 来自 org.springframework.kafka.support.serializer.JsonSerializer; 我已经看到有多个处理 JSON 对象的包（如 kafka.utils.json、com.google.gson.JsonObject; 等）。然后在运行时

ProducerRecord<Integer, AnyJson> record= new ProducerRecord<Integer, AnyJson>(topic, json);
kafkaProducer.send(record);

我有以下异常 找不到类 com.satori.rtm.connection.GsonSerializer$JsonElementWrapper 的序列化程序

有什么帮助吗？

【问题讨论】：

你可以使用 String SerDes，当然。但是你看到了什么问题？请edit 您的问题包括代码为minimal reproducible example 和错误。例如，您使用的是KafkaProducer<AnyJson, AnyJson>吗？
@cricket_007 我已经更新了问题
您需要将 Jackson 库与 Spring Kafka 包一起使用。 github.com/spring-projects/spring-kafka/blob/master/…
我有spring的依赖` org.springframework.kafkaspring-kafka2.1.6.RELEASE版本> org.springframework.bootspring-boot-starter-web2.0. 2.RELEASE`
是的，我知道。我指出你需要 Jackson 来处理 JSON，而不是 Gson，这是 com.satori.rtm.model.AnyJson 正在使用的

标签： json serialization apache-kafka

【解决方案1】：

您可能希望将org.apache.kafka.common.serialization.ByteArraySerializer 用于键和值序列化。现在您应该使用 byte[] 配置您的生产者记录。

然后使用来自 Jackson (http://www.baeldung.com/jackson-object-mapper-tutorial) 的 ObjectMapper 将任何 json 对象转换为字节数组。

【讨论】：