Spring Boot Kafka，消息头中的包名答案

【问题标题】：Spring Boot Kafka, package name in the message headerSpring Boot Kafka，消息头中的包名
【发布时间】：2022-01-13 15:04:17
【问题描述】：

我有两个服务eshop 和eshop-notification-service。服务 eshop 向 Kafka 发送消息：Notification(String message, LocalDateTime timestamp)，eshop-notification-service 使用它。

由于某种原因，在消息反序列化过程中，来自生产者的包名（com.example.eshop.model.Notification）保存在消息头中，结果，消费者端的反序列化失败并出现错误：

Caused by: org.springframework.messaging.converter.MessageConversionException: failed to resolve class name. Class not found [com.example.eshop.model.Notification]; nested exception is java.lang.ClassNotFoundException: com.example.eshop.model.Notification

而消费者端正确的包路径是：com.example.eshopnotification

我该如何解决这个问题？为什么包会保存在邮件头中？

【问题讨论】：

标签： spring-boot spring-kafka

【解决方案1】：

看到这个答案 - How to deserialize message in consumer when produced message is a model type which consumer has no reference to?

您可以使用类型映射，也可以将USE_TYPE_INFO_HEADERS 属性设置为 false 并设置默认类型。

你也可以在生产者端设置：

JsonSerializer.ADD_TYPE_INFO_HEADERS（默认为 true）：您可以将其设置为 false 以禁用 JsonSerializer 上的此功能（设置 addTypeInfo 属性）。

https://docs.spring.io/spring-kafka/docs/2.7.x/reference/html/#json-serde

【讨论】：

谢谢，这是正确的解决方案
但是如果我们的应用程序中有很多 @KafkaListeners 会发生什么，这里的常用方法是什么？
将反序列化移动到侦听器适配器 - 这样，转换器可以从方法签名中推断出类型。使用ByteArrayDeserializer 和JsonMessageConverter。请参阅docs.spring.io/spring-kafka/docs/current/reference/html/… 和此示例github.com/spring-projects/spring-kafka/tree/main/samples/…，它发送Foo1 并接收Foo2，其类型在侦听器方法中定义。只需将转换器添加为@Bean，Boot 就会将其连接到容器工厂。

【解决方案2】：

在 application.yml 中尝试如下映射配置

spring: 
  kafka:
    consumer:
      properties:
        spring:
          json:
            type:
              mapping: "com.example.eshop.model.Notification:com.example.eshopnotification.Notification"

【讨论】：

虽然这可行，但通常情况下，我们在生产者端从一个类映射到一个令牌（例如notification），在消费者端从令牌映射到一个类。这样，消费者就不必“知道”生产者类的名称；他们只需要就令牌达成一致。