如何覆盖Spring Cloud Stream的Jackson ObjectMapper配置？

问题根本原因 #

Spring Cloud Stream Kafka binder默认使用的ObjectMapper是其初始化内部MappingJackson2MessageConverter时自行构建的，不会复用Spring Boot全局配置的ObjectMapper，也不会拾取普通的MappingJackson2MessageConverter Bean，因此你之前的两个配置都没有作用到Kafka消息的反序列化流程中。

注意：请确保项目中已引入jackson-datatype-jsr310依赖（Spring Boot spring-boot-starter-json 已默认包含该依赖），否则JDK8时间类型的序列化反序列化会直接抛出异常。

自定义配置方案 #

场景1：使用@StreamListener注解式消费模型

直接定义名为kafkaMessageConverter的Bean，Kafka binder会优先使用这个Bean作为默认消息转换器，在里面传入你自定义配置的ObjectMapper即可：

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.springframework.context.annotation.Bean;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.kafka.support.converter.KafkaMessageConverter;
import org.springframework.kafka.support.converter.MappingJackson2MessageConverter;

@Bean
public KafkaMessageConverter kafkaMessageConverter() {
    // 自定义ObjectMapper配置，关闭时区自动调整，注册JDK8时间类型序列化模块
    ObjectMapper customObjectMapper = Jackson2ObjectMapperBuilder.json()
            .featuresToDisable(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE)
            .modules(new JavaTimeModule())
            // 其他你需要的Jackson配置
            .build();
    return new MappingJackson2MessageConverter(customObjectMapper);
}

场景2：使用函数式编程消费模型（Spring Cloud Stream 2020.0+ 版本默认）

如果你开启了原生解码配置spring.cloud.stream.kafka.bindings.<你的消费binding名称>.consumer.use-native-decoding=true，则需要自定义Kafka的Json反序列化器：

1. 首先添加自定义的Json反序列化器Bean：

import org.springframework.kafka.support.serializer.JsonDeserializer;

@Bean
public JsonDeserializer<你的消息POJO类型> customJsonDeserializer() {
    ObjectMapper customObjectMapper = Jackson2ObjectMapperBuilder.json()
            .featuresToDisable(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE)
            .modules(new JavaTimeModule())
            .build();
    return new JsonDeserializer<>(你的消息POJO类型.class, customObjectMapper);
}

2. 绑定配置文件中指定使用自定义反序列化器：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          <你的消费binding名称>:
            consumer:
              configuration:
                value.deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
                spring.json.trusted.packages: "你允许反序列化的包路径，生产环境不要配置为*"

内容的提问来源于stack exchange，提问作者E-Riz