在Kafka中生产和消费JSON

Question

我们将在项目中部署Apache Kafka2.10，并通过JSON对象在生产者和消费者之间进行通信。

到目前为止，我想我需要：

1. 实现自定义序列化程序以将JSON转换为字节数组
2. 实现自定义反序列化器，将字节数组转换为JSON对象
3. 生成消息
4. 读取消费者类中的消息

@Override
public byte[] serialize(String topic, T data) {
    if (data == null)
        return null;
    try {
        ObjectMapper mapper = new ObjectMapper();
        return mapper.writeValueAsBytes(data);
    } catch (Exception e) {
        throw new SerializationException("Error serializing JSON message", e);
    }
}

然而，到目前为止，有问题的2-4分。我在自定义反序列化程序中尝试了这样的操作：

@Override
public JsonNode deserialize(String topic, byte[] bytes) {
    if (bytes == null)
        return null;

    JsonNode data;
    try {
        objectMapper = new ObjectMapper();
        data = objectMapper.readTree(bytes);
    } catch (Exception e) {
        throw new SerializationException(e);
    }
    return data;
}

在我的消费者身上，我试着：

    KafkaConsumer<String, TextNode> consumer = new KafkaConsumer<String, TextNode>(messageConsumer.properties);
    consumer.subscribe(Arrays.asList(messageConsumer.topicName));
    int i = 0;
    while (true) {
        ConsumerRecords<String, TextNode> records = consumer.poll(100);
        for (ConsumerRecord<String, TextNode> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value().asText());
        }
    }

我以为这将生成一个正确的原始json字符串，但我调用record.value().aStext()得到的只是一些散列字符串“IntCIMTLEVWiIDOGXCJ2YWX1ZVWiIH0I”。

在Kafka中通过JSON进行通信的任何建议或示例都将非常感谢。

Answer 1

我建议您使用UTF-8编码作为字符串JSON序列化器:
1。Producer以JSON字符串(“{\”key\“:\”value\“}”)
2的形式获取数据。生产者使用UTF-8(jsonString.getBytes(StandardCharSets.utf_8)；)
3将JSON字符串序列化为字节。制作人将这些字节发送给Kafka4。消费者从Kafka5读取字节。使用者使用UTF-8将字节反序列化为JSON字符串(新字符串(consumedByteArray,StandardCharsets.utf_8)；)
6。使用者对JSON字符串执行任何需要的操作

我故意不使用您的代码，这样流程就可以理解了，我认为您可以很容易地将这个示例应用到您的项目中：）