Spring Boot中如何使用泛型类实现Kafka的JSON消息收发

一、修改Kafka生产者配置 #

将原配置中的泛型从TestClass替换为Response<TestClass>，同时配置JSON序列化的类型映射，确保泛型类能被正确序列化：

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
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.ibil.ocpi22.utils.kafka.dto.Response;
import com.ibil.ocpi22.utils.kafka.dto.TestClass;

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

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Response<TestClass>> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9094");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        // 配置类型映射，让序列化器识别Response类
        config.put(JsonSerializer.TYPE_MAPPINGS, "response:com.ibil.ocpi22.utils.kafka.dto.Response");

        return new DefaultKafkaProducerFactory<>(config);
    }

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

二、修改Kafka消费者配置 #

同样替换泛型为Response<TestClass>，并配置反序列化的信任包和类型映射，避免安全限制和类型识别错误：

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

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 com.ibil.ocpi22.utils.kafka.dto.Response;
import com.ibil.ocpi22.utils.kafka.dto.TestClass;

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, Response<TestClass>> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9094");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        // 允许反序列化指定包下的类
        config.put(JsonDeserializer.TRUSTED_PACKAGES, "com.ibil.ocpi22.utils.kafka.dto");
        // 和生产者保持一致的类型映射
        config.put(JsonDeserializer.TYPE_MAPPINGS, "response:com.ibil.ocpi22.utils.kafka.dto.Response");

        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(),
                new JsonDeserializer<>(Response.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Response<TestClass>> listener() {
        ConcurrentKafkaListenerContainerFactory<String, Response<TestClass>> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

三、生产消息示例 #

构造Response<TestClass>对象并发送到Kafka：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import com.ibil.ocpi22.utils.kafka.dto.Response;
import com.ibil.ocpi22.utils.kafka.dto.TestClass;

import java.time.Instant;

@Service
public class KafkaMessageProducer {

    private final KafkaTemplate<String, Response<TestClass>> kafkaTemplate;
    private static final String TOPIC = "test-topic";

    public KafkaMessageProducer(KafkaTemplate<String, Response<TestClass>> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage() {
        TestClass testData = TestClass.builder()
                .token("special_token_text")
                .url("SpecialUrl")
                .build();
        
        Response<TestClass> response = Response.<TestClass>builder()
                .data(testData)
                .timestamp(Instant.now())
                .build();
        
        kafkaTemplate.send(TOPIC, response);
    }
}

四、消费消息示例 #

监听Kafka主题，接收并处理Response<TestClass>类型的消息：

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

import com.ibil.ocpi22.utils.kafka.dto.Response;
import com.ibil.ocpi22.utils.kafka.dto.TestClass;

@Service
public class KafkaMessageConsumer {

    @KafkaListener(topics = "test-topic", groupId = "test-consumer")
    public void consumeMessage(Response<TestClass> response) {
        System.out.println("收到消息：");
        System.out.println("Data: " + response.getData());
        System.out.println("Timestamp: " + response.getTimestamp());
        // 若需输出JSON格式，可使用ObjectMapper将response转为JSON字符串
    }
}

关键注意事项 #

• 类型映射一致性：生产者和消费者的TYPE_MAPPINGS配置必须完全一致，否则序列化/反序列化时无法识别泛型类。
• 信任包配置：消费者必须配置TRUSTED_PACKAGES包含Response和TestClass所在的包，否则会触发安全限制导致反序列化失败。
• Instant类型支持：Spring Kafka的JSON序列化器默认支持Instant类型，无需额外配置即可完成日期的序列化/反序列化。

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