将自定义 Java 对象发送到 Kafka 主题

serialization java apache-kafka

2022-09-03 12:33:11

我有我的自定义Java对象，并希望在内置序列化中利用JVM将其发送到Kafka主题，但序列化失败并出现以下错误

org.apache.kafka.common.errors.SerializationException：无法将类 com.spring.kafka.Payload 的值转换为在 value.serializer 中指定的类 org.apache.kafka.common.serialization.ByteArraySerializer

有效载荷.java

public class Payload implements Serializable {

    private static final long serialVersionUID = 123L;

    private String name="vinod";

    private int anInt = 5;

    private Double aDouble = new Double("5.0");

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAnInt() {
        return anInt;
    }

    public void setAnInt(int anInt) {
        this.anInt = anInt;
    }

    public Double getaDouble() {
        return aDouble;
    }

    public void setaDouble(Double aDouble) {
        this.aDouble = aDouble;
    }

}

在我创建生产者期间，我设置了以下属性

<entry key="key.serializer"
                       value="org.apache.kafka.common.serialization.ByteArraySerializer" />
                <entry key="value.serializer"
                       value="org.apache.kafka.common.serialization.ByteArraySerializer" />

我的发送调用如下所示

kafkaProducer.send(new ProducerRecord<String, Payload>("test", new Payload()));

通过生产者将自定义java对象发送到kafka主题的正确方法是什么？

答案 1

我们有2个选项，如下所示

1）如果我们打算将自定义java对象发送给生产者，我们需要创建一个序列化程序来实现org.apache.kafka.common.serialization.Serializer，并在创建生产者期间传递该序列化器类

下面的代码参考

public class PayloadSerializer implements org.apache.kafka.common.serialization.Serializer {

    public void configure(Map map, boolean b) {

    }

    public byte[] serialize(String s, Object o) {

       try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(baos);
            oos.writeObject(o);
            oos.close();
            byte[] b = baos.toByteArray();
            return b;
        } catch (IOException e) {
            return new byte[0];
        }
    }

    public void close() {

    }
}

并相应地设置值序列化程序

<entry key="value.serializer"
                       value="com.spring.kafka.PayloadSerializer" />

2）无需创建自定义序列化程序类。使用现有的 ByteArraySerializer，但在发送过程中遵循该过程

Java Object -> String （最好是 JSON represenation 而不是 toString）->byteArray

答案 2

由于您正在使用，因此您需要实例化一个byte[]生产者。ByteArraySerializer

Producer<byte[],byte[]> producer = new KafkaProducer<>(props);

然后在序列化或其他一些方法之后生成传递byte[]，例如，

producer.send(new ProducerRecord<byte[],byte[]>("test", new Payload().toString().getBytes()));

如果您只将 Payload 对象传递给生产者，那么最好将密钥序列化程序和值序列化程序作为您打算传递的任何内容，并且在读取时需要从该数据中读取。

最好使用 Serializable 和 ByteArraySerializer/ByteArrayDeserializer。