Spring Kafka：@KafkaListener泛型T在抽象实现中无法解析

问题描述 #

使用带泛型T的抽象类AbstractTestKafka实现@KafkaListener逻辑，子类TestKafkaImpl指定T为Cat类型时，JSON消息被反序列化为LinkedHashMap而非Cat，抛出类型转换异常：

Caused by: java.lang.ClassCastException: class java.util.LinkedHashMap cannot be cast to class com.demo.Cat (java.util.LinkedHashMap is in module java.base of loader 'bootstrap'; com.demo.Cat is in unnamed module of loader 'app')
    at com.spring.kafka.poc.test.TestKafkaImpl.processNew(TestKafkaImpl.java:13) ~[classes/:na]

原因分析 #

Java泛型擦除机制导致运行时无法直接获取抽象类中泛型T的实际类型，Spring Kafka的默认JsonMessageConverter无法识别目标类型，只能将JSON反序列化为LinkedHashMap。

解决方案 #

步骤1：在抽象类中捕获泛型实际类型 #

通过反射获取子类继承时指定的泛型参数类型，存储为Class<T>字段，供后续反序列化使用：

public abstract class AbstractTestKafka<T> {
    protected final Class<T> payloadType;

    @SuppressWarnings("unchecked")
    protected AbstractTestKafka() {
        Type genericSuperclass = getClass().getGenericSuperclass();
        if (genericSuperclass instanceof ParameterizedType) {
            ParameterizedType parameterizedType = (ParameterizedType) genericSuperclass;
            this.payloadType = (Class<T>) parameterizedType.getActualTypeArguments()[0];
        } else {
            throw new IllegalArgumentException("子类必须继承带泛型参数的AbstractTestKafka");
        }
    }

    // 修改listen方法的参数类型为List<Message<?>>
    @KafkaListener(id = "newConsumer",topics = "topic1", containerFactory = "kafkaListenerContainerFactory",autoStartup = "true")
    public void listen(List<Message<?>> messages, Acknowledgment ack){
        Flux.fromIterable(messages)
                .log()
                .map(message -> {
                    // 将原始payload转换为目标类型T
                    Object rawPayload = message.getPayload();
                    if (rawPayload instanceof LinkedHashMap) {
                        ObjectMapper objectMapper = new ObjectMapper();
                        T payload = objectMapper.convertValue(rawPayload, payloadType);
                        return MessageBuilder.createMessage(payload, message.getHeaders());
                    }
                    return message;
                })
                .flatMap(this::processNew)
                .subscribeOn(Schedulers.boundedElastic())
                .blockLast(Duration.ofMillis(6000));
        ack.acknowledge();
    }

    public abstract Mono<Object> processNew(Message<T> msg) ;
}

步骤2：保持子类实现不变 #

子类TestKafkaImpl无需额外修改，继承抽象类时指定Cat即可：

public class TestKafkaImpl extends AbstractTestKafka<Cat> {
    @Override
    public Mono<Object> processNew(Message<Cat> msg) {
        Cat cat = msg.getPayload();
        // 业务逻辑处理
        return Mono.empty();
    }
}

可选优化：自定义消息转换器 #

如果希望在容器层面统一处理泛型类型，可以自定义BatchMessagingMessageConverter，结合Jackson2JavaTypeMapper动态识别目标类型。不过上述反射捕获泛型的方式更灵活，适配不同子类的类型需求。

验证效果 #

修改完成后，Kafka消息会被正确反序列化为Cat类型，processNew方法中可以直接获取Cat对象，不会再抛出ClassCastException。

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