java - 在 KAFKA 中生成和使用 JSON

Question

我们将在我们的项目上部署 Apache Kafka 2.10，并通过 JSON 对象在生产者和消费者之间进行通信。

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

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

关于第一点，我认为应该是这样的:

@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);
    }
}

哪里T data可以作为字符串 "{\"key\" : \"value\"}" 传递.

但是现在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. 生产者获取 JSON 字符串形式的数据 ("{\"key\":\"value\"}")
2. Producer 使用 UTF-8 将 JSON 字符串序列化为字节 (jsonString.getBytes(StandardCharsets.UTF_8);)
3. Producer将此字节发送到Kafka4. 消费者从 Kafka 读取字节5. 消费者使用 UTF-8 将字节反序列化为 JSON 字符串 (new String(consumedByteArray, StandardCharsets.UTF_8);)
6. 消费者使用 JSON 字符串做任何需要的事情

我故意没有使用您的代码，因此流程是可以理解的，我认为您可以非常轻松地将这个示例应用到您的项目中:)