什么形式的消息适合向 Kafka 发送值？

Question

我目前有如下开发环境

• spring-开机2.X
• spring-kafka 2.5.5
• 卡夫卡版本 2.5.1
• Producer/Consumer Key Serializer/Deserializer = String(De)serializer.class
• Producer/Consumer值Serializer/Deserializer=Json(德)serializer.class

Producer发送Kafka消息时，通过ObjectMapper.writeValueAsString(customDto).

将自定义DTO(或POJO)转为JSON String类型

• ProducerFactory
• KafkaTemplate<字符串, 字符串>

ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, msg); // The message is a JSON String variable made with ObjectMapper.

ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(record);

而Consumer接收到JSON String消息后，通过ObjectMapper.readValue(message, CustomDto.class)转换为对应的DTO(或POJO)接收

• ConsumerFactory

For reference, there is not only one type of Custom DTO.

我想知道通过每次将自定义DTO转换为JSON String来发送和接收自定义DTO是否正确。

或者有没有办法直接发送和接收自定义DTO而不需要转换？我想使用泛型。怎么样？

Answer 1

它必须转换为 byte[]，但您可以使用框架提供的 (De)Serializers 而不是自己动手。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#json-serde

Answer 2

有一种更好的方法可以通过 Kafka 发送复杂类型，它包含一个可以自动序列化和反销毁的结构。

1. 使用 Avro 架构 https://avro.apache.org/
2. 添加 confluent 模式注册表来处理 serialization/deserialization 过程和验证。 confluent

看看这个 example 看看它有多干净和高效。