如何实现Spring-Kafka通用序列化与反序列化配置？最优方案是什么？

问题背景 #

之前的做法是将Kafka消息序列化为字节数组，在每个@KafkaListener中手动接收并解析，导致大量重复代码，需要更优雅的全局配置方案。

最佳解决方案：Spring Boot全局JSON序列化/反序列化配置 #

通过Spring Boot提供的DefaultKafkaConsumerFactoryCustomizer、DefaultKafkaProducerFactoryCustomizer以及自定义RecordMessageConverter，可以实现全局的JSON序列化与反序列化配置，无需在每个监听器中重复处理。

消费者全局反序列化配置

@Configuration
class ConsumerKafkaConfiguration {
    @Bean
    fun defaultKafkaConsumerFactoryCustomizer() = DefaultKafkaConsumerFactoryCustomizer {
        it.keyDeserializer = StringDeserializer()
        it.valueDeserializer = ByteArrayDeserializer()
    }

    @Bean
    fun messageConvertor(): RecordMessageConverter {
        val converter = StringJsonMessageConverter()
        val typeMapper = DefaultJackson2JavaTypeMapper()

        // 优先使用消息中的TYPE_ID字段确定反序列化目标类型
        typeMapper.typePrecedence = Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID
        // 添加可信包路径，规避Jackson反序列化安全风险
        typeMapper.addTrustedPackages(ResolverKafkaEvents.PACKAGE_GEN_EVENTS)
        // 配置类型映射：自定义类型ID与实体类的对应关系
        typeMapper.idClassMapping = ResolverKafkaEvents
            .resolveEvents
            .mapValues { Class.forName(it.value) }

        converter.typeMapper = typeMapper

        return converter
    }
}

核心逻辑说明：

• DefaultKafkaConsumerFactoryCustomizer用于全局统一配置消费者序列化器，这里指定key为字符串反序列化、value为字节数组反序列化，后续由RecordMessageConverter完成JSON到实体类的转换。
• StringJsonMessageConverter配合DefaultJackson2JavaTypeMapper实现自动类型转换，通过typePrecedence确保优先使用消息中的类型标识字段，避免类型歧义。

生产者全局序列化配置

@Configuration
class ProducerKafkaConfiguration {

    @Bean
    fun defaultKafkaProducerFactoryCustomizer() = DefaultKafkaProducerFactoryCustomizer { factory ->
        factory.keySerializer = StringSerializer()
        factory.valueSerializer = JsonSerializer<Any>().apply {
            configure(
                mutableMapOf(
                    JsonSerializer.TYPE_MAPPINGS to ResolverKafkaEvents
                        .resolveEvents
                        .asSequence()
                        .joinToString(", ") { "${it.key}:${it.value}" }
                ),
                false,
            )
        }
    }

    @Bean
    fun customAsyncProducer(
        kafkaTemplate: KafkaTemplate<String, Any>
    ) = KafkaJsonProducer(kafkaTemplate)
}

核心逻辑说明：

• DefaultKafkaProducerFactoryCustomizer全局配置生产者序列化器，key用字符串序列化，value通过JsonSerializer自动将对象转为JSON，并通过TYPE_MAPPINGS配置类型映射，确保消费者能正确识别消息类型。
• 自定义KafkaJsonProducer封装KafkaTemplate，提供更简洁的消息发送API，避免重复创建模板实例。

方案优势 #

• 全局统一处理：无需在每个@KafkaListener中手动解析字节数组，所有消息的序列化/反序列化逻辑由配置统一管控。
• 类型安全：通过类型映射配置，确保消息能精准转换为目标实体类，避免类型转换错误。
• 减少冗余代码：将序列化/反序列化逻辑集中到配置类，降低代码重复度，提升可维护性。

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