Kafka Streams：POJO 序列化/反序列化

Question

我们可以使用 Kafka Streams 中的哪些类/方法将 Java 对象序列化/反序列化为字节数组，反之亦然？以下链接提出了 ByteArrayOutputStream 和 ObjectOutputStream 的用法，但它们不是线程安全的。

Send Custom Java Objects to Kafka Topic

还有另一个使用 ObjectMapper 的选项，ObjectReader（用于线程安全），但这是从 POJO -> JSON -> bytearray 转换的。似乎这个选项是一个广泛的选项。想检查是否有直接的方法将对象转换为字节数组，反之亦然，这是线程安全的。请推荐

import org.apache.kafka.common.serialization.Serializer;
public class HouseSerializer<T> implements Serializer<T>{
    private Class<T> tClass;
    public HouseSerializer(){

    }

    @SuppressWarnings("unchecked")
    @Override
    public void configure(Map configs, boolean isKey) {
        tClass = (Class<T>) configs.get("POJOClass");       
    }

    @Override
    public void close() {
    }

    @Override
    public byte[] serialize(String topic, T data) {
        //Object serialization to be performed here
        return null;
    }
}

注意：Kafka 版本 - 0.10.1

Answer 1

想检查是否有直接的方法将对象转换为字节数组

如果可能，我建议您将Avro serialization 与 Confluent Schema Registry 一起使用，但不是必需的。 JSON 是一个很好的后备方案，但会占用更多“在线”空间，因此MsgPack 将是那里的替代方案。

See Avro code example here

以上示例使用avro-maven-plugin 从src/main/resources/avro 架构文件生成一个LogLine 类。

---

否则how to serialize your object into a byte array由你自己决定，比如一个String一般打包成

[(length of string) (UTF8 encoded bytes)]

虽然布尔值是单个 0 或 1 位

线程安全

我理解您的担忧，但您通常不会在线程之间共享反序列化数据。您为每个独立的消息发送/读取/处理一条消息。