如何配置Kafka密钥序列化器，对于有时密钥是长的，有时密钥是字符串的情况？

Question

我有kafka producer配置,在此之前,我将密钥作为字符串类型发送,并配置了密钥序列化器,如下所示,

configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ObjectSerializer.class.getName());

但是我们有一个要求，一些时间键可能也很长，所以在不创建另一个生产者配置的情况下，我们可以像ObjectSerializer那样配置上面的东西吗？有这样的礼物吗？

我的spring-kafka版本是2.1，所以我不能使用DelegatingSerializer，也不能升级项目中的版本。

更新1：

我创建了像下面这样的自定义序列化器，并将其配置为我的密钥序列化器，但是在发布了带有一些长值的消息作为密钥，并且当我看到密钥包含一些垃圾符号时，如果有任何错误，可以更正下面的代码吗。

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

import java.io.UnsupportedEncodingException;
import java.util.Map;

public class LongStringSerializer implements Serializer<Object> {

    private String encoding = "UTF8";

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        String propertyName = isKey ? "key.serializer.encoding" : "value.serializer.encoding";
        Object encodingValue = configs.get(propertyName);
        if (encodingValue == null)
            encodingValue = configs.get("serializer.encoding");
        if (encodingValue != null && encodingValue instanceof String)
            encoding = (String) encodingValue;
    }

    @Override
    public byte[] serialize(String topic, Object data) {
        if (data == null)
            return null;

        if (data instanceof Long) {
            return serialize(Long.parseLong(data.toString()));
        }

        if (data instanceof String) {
            return serializeStringData((String) data);
        }
        return new byte[0];
    }

    private byte[] serializeStringData(String data) {
        try {
            return data.getBytes(encoding);
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding);
        }
    }

    private byte[] serialize(Long data) {
        if (data == null)
            return null;

        return new byte[] {
                (byte) (data >>> 56),
                (byte) (data >>> 48),
                (byte) (data >>> 40),
                (byte) (data >>> 32),
                (byte) (data >>> 24),
                (byte) (data >>> 16),
                (byte) (data >>> 8),
                data.byteValue()
        };
    }

    @Override
    public void close() {
        // nothing to do
    }
}

Answer 1

请参阅委派序列化程序。

版本2.3引入了DelegatingSerializer和DelegatingDeserializer，它们允许生成和使用具有不同键和/或值类型的记录。生产者必须将头DelegatingSerializer.Serialization_Selector设置为用于选择要使用的序列化程序的选择器值；如果未找到匹配项，则引发IllegalStateException。

它需要一个头来告诉它使用哪个序列化器，但是您可以编写一个简单的版本，只查看键类型。

（我想我将增强delegatingserializer来实现这一点）。

另一种选择是使用不同的生产者为每一个。

在2.5版中，您可以覆盖Kafkatemplate上的生产者工厂属性：

java prettyprint-override">    /**
     * Create an instance using the supplied producer factory and properties, with
     * autoFlush false. If the configOverrides is not null or empty, a new
     * {@link DefaultKafkaProducerFactory} will be created with merged producer properties
     * with the overrides being applied after the supplied factory's properties.
     * @param producerFactory the producer factory.
     * @param configOverrides producer configuration properties to override.
     * @since 2.5
     */
    public KafkaTemplate(ProducerFactory<K, V> producerFactory, @Nullable Map<String, Object> configOverrides) {