如何用Jackson将整数数组反序列化为LocalDateTime及配置Kafka反序列化

一、为什么默认JSR310模块不生效？ #

jackson-datatype-jsr310默认仅支持ISO-8601字符串格式（如"2022-09-21T21:28:37.9999999"）或时间戳数值格式，完全不兼容你这种[年,月,日,时,分,秒,纳秒]的整数数组结构，因此必须自定义反序列化逻辑。

二、自定义LocalDateTime反序列化器 #

1. 编写反序列化器实现类 #

继承JsonDeserializer<LocalDateTime>，实现数组到LocalDateTime的转换逻辑：

import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonDeserializer;
import java.io.IOException;
import java.time.LocalDateTime;

public class LocalDateTimeArrayDeserializer extends JsonDeserializer<LocalDateTime> {

    @Override
    public LocalDateTime deserialize(JsonParser parser, DeserializationContext ctx) throws IOException {
        Integer[] dateArray = parser.readValueAs(Integer[].class);
        if (dateArray == null || dateArray.length < 6) {
            throw ctx.mappingException("日期数组格式无效，至少需包含年、月、日、时、分、秒六个元素");
        }

        int year = dateArray[0];
        int month = dateArray[1];
        int day = dateArray[2];
        int hour = dateArray[3];
        int minute = dateArray[4];
        int second = dateArray[5];
        int nano = dateArray.length >=7 ? dateArray[6] : 0;

        return LocalDateTime.of(year, month, day, hour, minute, second, nano);
    }
}

2. 注册反序列化器 #

有两种注册方式可选：

方式一：全局注册（应用级生效）

在Spring配置类中配置自定义ObjectMapper：

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class JacksonConfig {

    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper objectMapper = new ObjectMapper();
        JavaTimeModule javaTimeModule = new JavaTimeModule();
        // 注册自定义反序列化器
        javaTimeModule.addDeserializer(LocalDateTime.class, new LocalDateTimeArrayDeserializer());
        objectMapper.registerModule(javaTimeModule);
        return objectMapper;
    }
}

方式二：字段级注册（仅特定字段生效）

直接在实体类的date字段上标注注解：

import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import java.time.LocalDateTime;

public class KafkaMessage {

    @JsonDeserialize(using = LocalDateTimeArrayDeserializer.class)
    private LocalDateTime date;

    // getter、setter方法省略
}

三、Spring Kafka（含嵌入式Kafka）集成JSON反序列化器 #

如果不想手动调用ObjectMapper做映射，可以直接配置Spring Kafka的JsonDeserializer，让框架自动完成消息反序列化：

1. 配置消费者工厂 #

在Spring配置类中定义消费者工厂，注入带自定义反序列化器的ObjectMapper：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, KafkaMessage> consumerFactory(ObjectMapper objectMapper) {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 嵌入式Kafka可替换为对应地址
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "your-group-id");
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        // 初始化带自定义ObjectMapper的JsonDeserializer
        JsonDeserializer<KafkaMessage> jsonDeserializer = new JsonDeserializer<>(KafkaMessage.class, objectMapper);
        jsonDeserializer.addTrustedPackages("*"); // 按需调整信任包范围

        return new DefaultKafkaConsumerFactory<>(configProps, new StringDeserializer(), jsonDeserializer);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, KafkaMessage> kafkaListenerContainerFactory(ConsumerFactory<String, KafkaMessage> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, KafkaMessage> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }
}

2. 嵌入式Kafka适配 #

如果用嵌入式Kafka做测试，只需在测试类中加载上述配置即可，无需额外修改：

import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = "your-topic")
public class KafkaConsumerTest {
    // 测试逻辑省略
}

3. 直接消费消息 #

配置完成后，可在@KafkaListener方法中直接接收KafkaMessage对象，无需手动处理反序列化：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageConsumer {

    @KafkaListener(topics = "your-topic", groupId = "your-group-id")
    public void consume(KafkaMessage message) {
        LocalDateTime date = message.getDate();
        // 业务逻辑处理
    }
}

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