如何在Spring Kafka的JsonSerializer中注入ObjectMapper Bean？

问题根源 #

Spring Kafka默认在AbstractConfig类中通过无参构造方法实例化ValueSerializer，默认创建的JsonSerializer使用独立的默认ObjectMapper实例，不会读取Spring上下文里绑定了spring.jackson配置的全局ObjectMapper Bean，因此你在yml中添加的spring.jackson.default-property-inclusion: NON_EMPTY配置不会生效。

可行解决方案 #

方案1：自定义ProducerFactory手动注入ObjectMapper（推荐） #

该方案不需要额外创建自定义类，直接复用Spring上下文已配置好的ObjectMapper，同时保留所有原生kafka配置的读取逻辑：

• 直接注入Spring自动配置的、已经加载了yml中Jackson配置的ObjectMapper Bean
• 手动实例化JsonSerializer时传入该ObjectMapper，绑定到生产者工厂
  代码示例：

import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.StringSerializer;

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory(
            KafkaProperties kafkaProperties,
            ObjectMapper objectMapper) {
        // 读取application.yml中所有spring.kafka.producer开头的原生配置
        var producerConfigs = kafkaProperties.buildProducerProperties();
        // 用预配置的ObjectMapper创建JsonSerializer
        JsonSerializer<Object> valueSerializer = new JsonSerializer<>(objectMapper);
        // 可按需配置JsonSerializer的其他属性，比如是否添加类型头
        valueSerializer.setAddTypeInfo(true);
        return new DefaultKafkaProducerFactory<>(
                producerConfigs,
                new StringSerializer(),
                valueSerializer
        );
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}

方案2：自定义JsonSerializer实现类 #

如果希望通过配置文件指定序列化器，不需要手动配置ProducerFactory，可以自定义序列化器子类：

1. 创建自定义JsonSerializer类，通过构造方法注入Spring托管的ObjectMapper

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.serializer.JsonSerializer;

public class CustomJacksonJsonSerializer extends JsonSerializer<Object> {
    @Autowired
    public CustomJacksonJsonSerializer(ObjectMapper objectMapper) {
        super(objectMapper);
    }
}

2. 在application.yml中指定value序列化器为自定义类

spring:
  kafka:
    producer:
      value-serializer: com.yourproject.package.CustomJacksonJsonSerializer

如果需要对消费者的JsonDeserializer做相同的自定义配置，逻辑完全一致，替换对应反序列化器即可

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