编写自定义Kafka序列化程序

Question

我在一个Kafka消息中使用了我自己的类，它有一堆字符串数据类型。

我想我需要编写自己的序列化器并将其提供给生产者属性？

Answer 1

编辑

在较新的Kafka客户端中，实现serializer而不是encoder。

编写自定义序列化程序所需的内容有：

null

如果遵循Kafka的Producer示例，您将通过Properties对象构造ProducerConfig。在构建属性文件时，请确保包括：

props.put("serializer.class", "path.to.your.CustomSerializer");

带有要让Kafka在将消息追加到日志之前用于序列化消息的类的路径。

编写Kafka可以正确解释的自定义序列化程序需要实现Kafka提供的encoder[T]scala类。在java中实现traits很奇怪，但在我的项目中序列化JSON的方法如下：

public class JsonEncoder implements Encoder<Object> {
    private static final Logger logger = Logger.getLogger(JsonEncoder.class);
    // instantiating ObjectMapper is expensive. In real life, prefer injecting the value.
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public JsonEncoder(VerifiableProperties verifiableProperties) {
        /* This constructor must be present for successful compile. */
    }

    @Override
    public byte[] toBytes(Object object) {
        try {
            return objectMapper.writeValueAsString(object).getBytes();
        } catch (JsonProcessingException e) {
            logger.error(String.format("Json processing failed for object: %s", object.getClass().getName()), e);
        }
        return "".getBytes();
    }
}

您的问题听起来像是在为添加到日志中的所有消息使用一个对象（我们称之为customMessage)。如果是这种情况,那么序列化程序可能更像这样：

package com.project.serializer;
    
public class CustomMessageEncoder implements Encoder<CustomMessage> {
    public CustomMessageEncoder(VerifiableProperties verifiableProperties) {
        /* This constructor must be present for successful compile. */
    }

    @Override
    public byte[] toBytes(CustomMessage customMessage) {
        return customMessage.toBytes();
    }
}

这将使您的属性配置如下所示：

props.put("serializer.class", "path.to.your.CustomSerializer");