如何将LinkedHashMap转为自定义对象？Kafka消费类型转换异常排查

问题 #

我的Kafka消费者同时监听topic-a和topic-b两个主题，消息都指向同一个处理方法。相关代码如下：

消费者服务类 #

@Slf4j
@Service
public class KafkaConsumerService {

    @KafkaListener(topics = { "topic-a", "topic-b" }, groupId = "X-consumer")
    public <T> void listenX(Response<T> msg) {
        log.info("MSG arrived to consumer");
        KafkaMessageDTO<TokenMobieDTO> test = msg.getData();
        // 尝试提取object时抛出类型转换异常
    }
}

Kafka消息DTO类 #

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class KafkaMessageDTO<T> {

    @JsonProperty("object")
    private T object;

    @JsonProperty("header")
    private HttpHeaders headers;

}

我发送的消息类型都是KafkaMessageDTO<TokenMobieDTO>，调试时能看到数据结构完全匹配，但提取object时抛出以下异常：

threw exception; nested exception is java.lang.ClassCastException: class java.util.LinkedHashMap cannot be cast to class com.token.dto.TokenMobieDTO (java.util.LinkedHashMap is in module java.base of loader 'bootstrap';

尝试过手动映射，但因为没有List结构无法迭代，问题仍未解决。请问如何正确将msg对象转换为KafkaMessageDTO<TokenMobieDTO>类型？

解决方案 #

这个问题的核心是Jackson反序列化泛型类型时，无法自动推断嵌套泛型的具体参数，只能将object字段默认反序列化为LinkedHashMap，导致类型转换失败。可以通过以下几种方式解决：

1. 直接指定具体泛型类型（最简方案） #

修改监听方法的参数类型，去掉泛型<T>，直接明确指定Response<KafkaMessageDTO<TokenMobieDTO>>，让Jackson清楚知道要反序列化的具体类型：

@KafkaListener(topics = { "topic-a", "topic-b" }, groupId = "X-consumer")
public void listenX(Response<KafkaMessageDTO<TokenMobieDTO>> msg) {
    log.info("MSG arrived to consumer");
    KafkaMessageDTO<TokenMobieDTO> test = msg.getData();
    TokenMobieDTO token = test.getObject(); // 此时可正常获取，无类型转换异常
}

2. 使用ParameterizedTypeReference手动转换（保留泛型灵活性） #

如果需要保留方法的泛型特性，可借助ObjectMapper和ParameterizedTypeReference完成类型转换：

@Autowired
private ObjectMapper objectMapper;

@KafkaListener(topics = { "topic-a", "topic-b" }, groupId = "X-consumer")
public <T> void listenX(Response<T> msg) {
    log.info("MSG arrived to consumer");
    // 将msg.getData()转换为指定泛型类型
    KafkaMessageDTO<TokenMobieDTO> test = objectMapper.convertValue(
        msg.getData(),
        new ParameterizedTypeReference<KafkaMessageDTO<TokenMobieDTO>>() {}
    );
    TokenMobieDTO token = test.getObject();
}

3. 配置全局Kafka消息转换器（批量生效） #

通过配置MappingJackson2MessageConverter并开启类型信息，让消费者全局支持泛型反序列化：

@Configuration
public class KafkaConfig {

    @Bean
    public MappingJackson2MessageConverter messageConverter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setObjectMapper(objectMapper());
        return converter;
    }

    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper mapper = new ObjectMapper();
        // 开启默认类型写入，Jackson会在序列化时记录类型信息，反序列化时自动识别
        mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
        return mapper;
    }
}

注意：该方案要求生产者端使用相同的ObjectMapper配置，确保序列化时写入类型信息，否则消费者无法正确识别。

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