如何在Kafka中通过properties文件配置类型映射实现自定义消息收发

错误原因 #

1. spring.json.type.mapping同一端配置中，同一个别名customMessage绑定了多个不同包的全限定类名，序列化/反序列化器无法确定要加载的目标类
2. 消费者配置中逗号后存在多余空格，会被解析为类名的前缀，触发类找不到的报错

配置规则说明 #

spring.json.type.mapping采用别名:全限定类名的键值对格式，多个键值对用半角逗号分隔：

• 同一端同一个别名只能对应一个本地的全限定类名，不可重复绑定
• 键值对之间不要添加多余空格，避免被解析为类名的一部分
• 生产者与消费者的别名需保持一致：生产者序列化时会将别名写入Kafka消息头，消费者反序列化时通过别名匹配本地绑定的类

正确配置 #

生产者端application.properties #

# other properties ( I use CloudKarafka )
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
# 仅绑定生产者本地CustomMessage的全限定类名
spring.kafka.producer.properties.spring.json.type.mapping=customMessage:assignment2.kafka.CustomMessage

消费者端application.properties #

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
# 仅绑定消费者本地CustomMessage的全限定类名，去掉多余空格和重复绑定
spring.kafka.consumer.properties.spring.json.type.mapping=customMessage:ro.tuc.ds2020.kafkaconsumer.CustomMessage
# 可选配置，解决反序列化类信任校验报错，也可替换为自定义类所在的包名
spring.kafka.consumer.properties.spring.json.trusted.packages=*

补充说明 #

如果你的业务场景仅传递CustomMessage一种类型的消息，也可以省略类型映射配置，直接在消费者端指定默认反序列化目标类即可：

spring.kafka.consumer.properties.spring.json.value.default.type=ro.tuc.ds2020.kafkaconsumer.CustomMessage

同时需要保证两端CustomMessage的字段名、字段类型完全一致，避免反序列化时字段丢失。

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