如何通过配置动态切换Kafka Producer的消息序列化类型？

你当前通过复制KafkaSender Bean来切换JSON/字节数组序列化的方式确实冗余，以下是两种更优雅的实现方案，无需重复定义Bean：

方案1：自定义动态序列化器（推荐） #

核心是实现一个可根据目标主题自动切换的序列化器，将序列化逻辑统一封装，仅需单个KafkaSender Bean。

1.1 实现自定义序列化器 #

public class DynamicValueSerializer implements Serializer<Object> {

    private KafkaJsonSerializer jsonSerializer;
    private ByteArraySerializer byteArraySerializer;
    private Map<String, String> topicSerializerMapping;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 初始化内置序列化器
        jsonSerializer = new KafkaJsonSerializer();
        jsonSerializer.configure(configs, isKey);
        byteArraySerializer = new ByteArraySerializer();
        byteArraySerializer.configure(configs, isKey);
        // 读取主题-序列化器映射配置
        topicSerializerMapping = (Map<String, String>) configs.get("dynamic.serializer.topic.mapping");
    }

    @Override
    public byte[] serialize(String topic, Object data) {
        // 根据主题选择序列化方式
        String serializerType = topicSerializerMapping.getOrDefault(topic, "json");
        if ("bytearray".equals(serializerType)) {
            // 将SomePojo转为字节数组（可根据需求替换序列化方式，比如Protobuf）
            if (data instanceof SomePojo) {
                try {
                    return new ObjectMapper().writeValueAsBytes(data);
                } catch (JsonProcessingException e) {
                    throw new SerializationException("序列化SomePojo到字节数组失败", e);
                }
            }
            return byteArraySerializer.serialize(topic, data);
        } else {
            return jsonSerializer.serialize(topic, data);
        }
    }

    @Override
    public void close() {
        jsonSerializer.close();
        byteArraySerializer.close();
    }
}

1.2 配置单例KafkaSender Bean #

@Bean
public KafkaSender<String, Object> kafkaSender(
        @Value("${kafka.bootstrap-servers}") String bootstrapServers,
        @Value("#{${kafka.dynamic-serializer.topic-mapping}}") Map<String, String> topicSerializerMapping) {

    Map<String, Object> properties = new HashMap<>();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DynamicValueSerializer.class);
    // 传递主题-序列化器映射到自定义序列化器
    properties.put("dynamic.serializer.topic.mapping", topicSerializerMapping);
    // 配置JSON序列化器的类型映射（按需添加）
    properties.put(JsonSerializer.TYPE_MAPPINGS, "com.example.SomePojo:SomePojo");

    SenderOptions<String, Object> senderOptions = SenderOptions.create(properties);
    return KafkaSender.create(senderOptions);
}

1.3 配置文件中定义主题映射（application.yml） #

kafka:
  bootstrap-servers: your-bootstrap-server:9092
  dynamic-serializer:
    topic-mapping:
      topic-json: json          # 该主题用JSON序列化
      topic-bytearray: bytearray # 该主题用字节数组序列化

1.4 发送消息 #

只需指定目标主题，序列化器会自动处理：

// 发送JSON到topic-json
kafkaSender.send(SenderRecord.create("topic-json", "key", somePojo)).subscribe();

// 发送字节数组到topic-bytearray
kafkaSender.send(SenderRecord.create("topic-bytearray", "key", somePojo)).subscribe();

方案2：用工厂类动态创建KafkaSender #

如果需要明确控制序列化方式，可封装一个工厂类，根据需求动态生成不同配置的KafkaSender，避免重复Bean定义。

2.1 实现KafkaSender工厂 #

@Component
public class KafkaSenderFactory {

    private final String bootstrapServers;

    public KafkaSenderFactory(@Value("${kafka.bootstrap-servers}") String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }

    public <T> KafkaSender<String, T> getSender(Class<? extends Serializer<T>> valueSerializerClass) {
        Map<String, Object> properties = new HashMap<>();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializerClass);

        SenderOptions<String, T> senderOptions = SenderOptions.create(properties);
        return KafkaSender.create(senderOptions);
    }
}

2.2 使用工厂获取Sender #

// 获取JSON序列化的Sender
KafkaSender<String, SomePojo> jsonSender = kafkaSenderFactory.getSender(KafkaJsonSerializer.class);
jsonSender.send(SenderRecord.create("topic-json", "key", somePojo)).subscribe();

// 获取字节数组序列化的Sender
KafkaSender<String, byte[]> byteArraySender = kafkaSenderFactory.getSender(ByteArraySerializer.class);
// 手动将SomePojo转为字节数组
byte[] data = new ObjectMapper().writeValueAsBytes(somePojo);
byteArraySender.send(SenderRecord.create("topic-bytearray", "key", data)).subscribe();

关于你设想的动态泛型Bean #

Java泛型是编译时类型，运行时会被类型擦除，无法直接实现你设想的{retrieve from some configuration}动态泛型类型。上述两种方案已经覆盖了动态切换序列化的需求，是实际项目中常用的优雅实现方式。

内容的提问来源于stack exchange，提问作者PatPanda