序列化、反序列化与消息转换

Doubao 中英对照 Serialization, Deserialization, and Message Conversion

概述

Apache Kafka 提供了一个高级 API，用于序列化和反序列化记录值及其键。它以 org.apache.kafka.common.serialization.Serializer<T> 与 org.apache.kafka.common.serialization.Deserializer<T> 抽象的形式存在，并附带一些内置实现。同时，我们可以通过使用 Producer 或 Consumer 配置属性来指定序列化器和反序列化器类。以下示例展示了如何做到这一点：

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
...
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

对于更复杂或特殊的情况，KafkaConsumer（以及因此 KafkaProducer）提供了重载构造函数，分别为 keys 和 values 接收 Serializer 和 Deserializer 实例。

当你使用此 API 时，DefaultKafkaProducerFactory 和 DefaultKafkaConsumerFactory 也提供属性（通过构造函数或 setter 方法），将自定义 Serializer 和 Deserializer 实例注入目标 Producer 或 Consumer。此外，你可以通过构造函数传入 Supplier<Serializer> 或 Supplier<Deserializer> 实例——这些 Supplier 在每个 Producer 或 Consumer 创建时被调用。

字符串序列化

自 2.5 版本起，Spring for Apache Kafka 提供了 ToStringSerializer 和 ParseStringDeserializer 类，它们使用实体的字符串表示。它们依赖 toString 方法以及某个 Function<String> 或 BiFunction<String, Headers> 来解析字符串并填充实例的属性。通常，这会调用类上的某个静态方法，例如 parse：

ToStringSerializer<Thing> thingSerializer = new ToStringSerializer<>();
//...
ParseStringDeserializer<Thing> deserializer = new ParseStringDeserializer<>(Thing::parse);

默认情况下，ToStringSerializer 被配置为在记录 Headers 中传达有关被序列化实体的类型信息。你可以通过将 addTypeInfo 属性设为 false 来禁用此行为。接收侧的 ParseStringDeserializer 可以使用此信息。

• ToStringSerializer.ADD_TYPE_INFO_HEADERS（默认 true）：你可以将其设为 false 以在 ToStringSerializer 上禁用此特性（设置 addTypeInfo 属性）。

ParseStringDeserializer<Object> deserializer = new ParseStringDeserializer<>((str, headers) -> {
    byte[] header = headers.lastHeader(ToStringSerializer.VALUE_TYPE).value();
    String entityType = new String(header);

    if (entityType.contains("Thing")) {
        return Thing.parse(str);
    }
    else {
        // ...parsing logic
    }
});

你可以配置用于在 String 与 byte[] 之间转换的 Charset，默认为 UTF-8。

你可以使用 ConsumerConfig 属性配置反序列化器所使用的解析方法名：

• ParseStringDeserializer.KEY_PARSER
• ParseStringDeserializer.VALUE_PARSER

这些属性必须包含类的全限定名，后接方法名，以句点 . 分隔。该方法必须是静态的，签名为 (String, Headers) 或 (String) 之一。

还提供了一个 ToFromStringSerde，用于 Kafka Streams。

JSON

Spring for Apache Kafka 还提供了基于 Jackson JSON 对象映射器的 JacksonJsonSerializer 和 JacksonJsonDeserializer 实现。JacksonJsonSerializer 允许把任意 Java 对象写为 JSON byte[]。JacksonJsonDeserializer 需要一个额外的 Class<?> targetType 参数，以便将消费到的 byte[] 反序列化为适当的目标对象。以下示例展示了如何创建 JacksonJsonDeserializer：

JacksonJsonDeserializer<Thing> thingDeserializer = new JacksonJsonDeserializer<>(Thing.class);

你可以用一个 ObjectMapper 同时定制 JacksonJsonSerializer 和 JacksonJsonDeserializer。你也可以扩展它们，在 configure(Map<String, ?> configs, boolean isKey) 方法中实现某些特定的配置逻辑。

从 2.3 版本开始，所有感知 JSON 的组件默认都用 JacksonUtils.enhancedObjectMapper() 实例配置，该实例禁用了 MapperFeature.DEFAULT_VIEW_INCLUSION 和 DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES 特性。此外，该实例还配备了用于自定义数据类型的知名模块，例如 Java 时间和 Kotlin 支持。更多信息请参阅 JacksonUtils.enhancedObjectMapper() JavaDocs。该方法还会为 org.springframework.util.MimeType 对象的序列化注册一个 org.springframework.kafka.support.JacksonMimeTypeModule，将其序列化为纯字符串，以便网络上跨平台兼容。JacksonMimeTypeModule 可以在应用上下文中注册为一个 bean，它将被自动配置进 Spring Boot ObjectMapper 实例。

同样从 2.3 版本开始，JsonDeserializer 提供了基于 TypeReference 的构造函数，以便更好地处理目标泛型容器类型。

从 2.1 版本开始，你可以在记录 Headers 中传达类型信息，从而支持多种类型。此外，你可以使用以下 Kafka 属性配置序列化器和反序列化器。如果你已分别为 KafkaConsumer 和 KafkaProducer 提供了 Serializer 和 Deserializer 实例，这些属性不生效。

配置属性

• JacksonJsonSerializer.ADD_TYPE_INFO_HEADERS（默认 true）：你可以将其设为 false 以在 JacksonJsonSerializer 上禁用此特性（设置 addTypeInfo 属性）。
• JacksonJsonSerializer.TYPE_MAPPINGS（默认 empty）：见 映射类型。
• JacksonJsonDeserializer.USE_TYPE_INFO_HEADERS（默认 true）：你可以将其设为 false 以忽略序列化器设置的头。
• JacksonJsonDeserializer.REMOVE_TYPE_INFO_HEADERS（默认 true）：你可以将其设为 false 以保留序列化器设置的头。
• JacksonJsonDeserializer.KEY_DEFAULT_TYPE：当不存在头信息时，用于键反序列化的回退类型。
• JacksonJsonDeserializer.VALUE_DEFAULT_TYPE：当不存在头信息时，用于值反序列化的回退类型。
• JacksonJsonDeserializer.TRUSTED_PACKAGES（默认 java.util、java.lang）：允许反序列化的包模式列表，以逗号分隔。* 表示反序列化所有。
• JacksonJsonDeserializer.TYPE_MAPPINGS（默认 empty）：见 映射类型。
• JacksonJsonDeserializer.KEY_TYPE_METHOD（默认 empty）：见 使用方法确定类型。
• JacksonJsonDeserializer.VALUE_TYPE_METHOD（默认 empty）：见 使用方法确定类型。

从 2.2 版本开始，类型信息头（如果由序列化器添加）会被反序列化器移除。你可以通过将 removeTypeHeaders 属性设为 false 来恢复此前行为，可以直接在反序列化器上设置，或使用前文所述的配置属性。

另请参阅 定制 JacksonJsonSerializer 和 JacksonJsonDeserializer。

important

从 2.8 版本开始，如果你按 编程式构造 所示以编程方式构造序列化器或反序列化器，上述属性将由工厂应用，只要你尚未显式设置任何属性（使用 set*() 方法或流式 API）。此前，以编程方式创建时，配置属性从不被应用；如果你直接在对象上显式设置属性，情况仍然如此。

映射类型

从 2.2 版本开始，使用 JSON 时，你现在可以通过使用前述列表中的属性来提供类型映射。此前，你必须在序列化器和反序列化器内部自定义类型映射器。映射由逗号分隔的 token:className 对列表组成。出站时，载荷的类名被映射到对应的 token。入站时，类型头中的 token 被映射到对应的类名。

以下示例创建一组映射：

senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
senderProps.put(JsonSerializer.TYPE_MAPPINGS, "cat:com.mycat.Cat, hat:com.myhat.Hat");
...
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
consumerProps.put(JsonDeserializer.TYPE_MAPPINGS, "cat:com.yourcat.Cat, hat:com.yourhat.Hat");

important

对应的对象必须兼容。

如果你使用 Spring Boot，你可以在 application.properties（或 yaml）文件中提供这些属性。以下示例展示了如何做到这一点：

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties.spring.json.type.mapping=cat:com.mycat.Cat,hat:com.myhat.Hat

important

你只能用属性执行简单配置。对于更高级的配置（例如在序列化器和反序列化器中使用自定义 ObjectMapper），你应使用接收预构建序列化器和反序列化器的生产者工厂和消费者工厂构造函数。以下 Spring Boot 示例覆盖了默认工厂：

@Bean
public ConsumerFactory<?, ?> kafkaConsumerFactory(JsonDeserializer customValueDeserializer) {
    Map<String, Object> properties = new HashMap<>();
    // properties.put(..., ...)
    // ...
    return new DefaultKafkaConsumerFactory<>(properties,
        new StringDeserializer(), customValueDeserializer);
}

@Bean
public ProducerFactory<?, ?> kafkaProducerFactory(JsonSerializer customValueSerializer) {
    return new DefaultKafkaProducerFactory<>(properties.buildProducerProperties(),
        new StringSerializer(), customValueSerializer);
}

也提供了 setter，作为使用这些构造函数的替代方式。
:::

备注

使用 Spring Boot 并按上文覆盖 ConsumerFactory 和 ProducerFactory 时，bean 方法返回类型需要使用通配泛型类型。如果改为提供具体泛型类型，Spring Boot 将忽略这些 bean 而仍使用默认的。

从 2.2 版本开始，你可以通过使用带布尔参数 useHeadersIfPresent（默认为 true）的重载构造函数之一，显式配置反序列化器使用所提供的目标类型并忽略头中的类型信息。以下示例展示了如何做到这一点：

DefaultKafkaConsumerFactory<Integer, Cat1> cf = new DefaultKafkaConsumerFactory<>(props,
        new IntegerDeserializer(), new JacksonJsonDeserializer<>(Cat1.class, false));

使用方法确定类型

从 2.5 版本开始，你现在可以通过属性配置反序列化器调用一个方法来确定目标类型。如果存在，这将覆盖上文讨论的任何其他技术。如果数据由一个不使用 Spring 序列化器的应用发布，而你需要根据数据或其他头反序列化为不同类型，这会很有用。将这些属性设为方法名——类的全限定名后接方法名，以句点 . 分隔。该方法必须声明为 public static，具有 (String topic, byte[] data, Headers headers)、(byte[] data, Headers headers) 或 (byte[] data) 三种签名之一，并返回一个 Jackson JavaType。

• JsonDeserializer.KEY_TYPE_METHOD：spring.json.key.type.method
• JsonDeserializer.VALUE_TYPE_METHOD：spring.json.value.type.method

你可以使用任意头或检查数据来确定类型。

JavaType thing1Type = TypeFactory.defaultInstance().constructType(Thing1.class);

JavaType thing2Type = TypeFactory.defaultInstance().constructType(Thing2.class);

public static JavaType thingOneOrThingTwo(byte[] data, Headers headers) {
    // {"thisIsAFieldInThing1":"value", ...
    if (data[21] == '1') {
        return thing1Type;
    }
    else {
        return thing2Type;
    }
}

对于更复杂的数据检查，可考虑使用 JsonPath 或类似工具，但确定类型的测试越简单，处理过程越高效。

以下是一个以编程方式创建反序列化器的示例（在构造函数中向消费者工厂提供反序列化器时）：

JacksonJsonDeserializer<Object> deser = new JacksonJsonDeserializer<>()
        .trustedPackages("*")
        .typeResolver(SomeClass::thing1Thing2JavaTypeForTopic);

...

public static JavaType thing1Thing2JavaTypeForTopic(String topic, byte[] data, Headers headers) {
    ...
}

编程式构造

当以编程方式构造序列化器/反序列化器以用于生产者/消费者工厂时，自 2.3 版本起，你可以使用流式 API，它简化了配置。

@Bean
public ProducerFactory<MyKeyType, MyValueType> pf() {
    Map<String, Object> props = new HashMap<>();
    // props.put(..., ...)
    // ...
    DefaultKafkaProducerFactory<MyKeyType, MyValueType> pf = new DefaultKafkaProducerFactory<>(props,
        new JacksonJsonSerializer<MyKeyType>()
            .forKeys()
            .noTypeInfo(),
        new JacksonJsonSerializer<MyValueType>()
            .noTypeInfo());
    return pf;
}

@Bean
public ConsumerFactory<MyKeyType, MyValueType> cf() {
    Map<String, Object> props = new HashMap<>();
    // props.put(..., ...)
    // ...
    DefaultKafkaConsumerFactory<MyKeyType, MyValueType> cf = new DefaultKafkaConsumerFactory<>(props,
        new JacksonJsonDeserializer<>(MyKeyType.class)
            .forKeys()
            .ignoreTypeHeaders(),
        new JacksonJsonDeserializer<>(MyValueType.class)
            .ignoreTypeHeaders());
    return cf;
}

要以编程方式提供类型映射（类似于 使用方法确定类型），使用 typeFunction 属性。

JacksonJsonDeserializer<Object> deser = new JacksonJsonDeserializer<>()
        .trustedPackages("*")
        .typeFunction(MyUtils::thingOneOrThingTwo);

或者，只要你不使用流式 API 配置属性、也不使用 set*() 方法设置它们，工厂就会使用配置属性配置序列化器/反序列化器；见 配置属性。

委托序列化器与反序列化器

使用头

2.3 版本引入了 DelegatingSerializer 和 DelegatingDeserializer，它们允许以不同的键和/或值类型生产和消费记录。生产者必须把一个头 DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR 设置为一个选择器值，用于为值选择使用哪个序列化器；并为键设置 DelegatingSerializer.KEY_SERIALIZATION_SELECTOR；如果找不到匹配，则抛出 IllegalStateException。

对于入站记录，反序列化器使用相同的头来选择要使用的反序列化器；如果找不到匹配或该头不存在，则返回原始 byte[]。

你可以通过构造函数配置从选择器到 Serializer / Deserializer 的映射，也可以通过 Kafka 生产者/消费者属性、使用键 DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR_CONFIG 和 DelegatingSerializer.KEY_SERIALIZATION_SELECTOR_CONFIG 来配置。对于序列化器，生产者属性可以是一个 Map<String, Object>，其中键是选择器，值是 Serializer 实例、序列化器 Class 或类名。该属性也可以是逗号分隔的映射条目组成的字符串，如下所示。

对于反序列化器，消费者属性可以是一个 Map<String, Object>，其中键是选择器，值是 Deserializer 实例、反序列化器 Class 或类名。该属性也可以是逗号分隔的映射条目组成的字符串，如下所示。

要使用属性进行配置，请使用以下语法：

producerProps.put(DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR_CONFIG,
    "thing1:com.example.MyThing1Serializer, thing2:com.example.MyThing2Serializer")

consumerProps.put(DelegatingDeserializer.VALUE_SERIALIZATION_SELECTOR_CONFIG,
    "thing1:com.example.MyThing1Deserializer, thing2:com.example.MyThing2Deserializer")

生产者随后会把 DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR 头设为 thing1 或 thing2。

此技术支持向同一主题（或不同主题）发送不同类型。

备注

从 2.5.1 版本开始，如果类型（键或值）是 Serdes 支持的标准类型之一（Long、Integer 等），则不必设置选择器头。相反，序列化器会把该头设为该类型的类名。无需为这些类型配置序列化器或反序列化器，它们将被（一次性）动态创建。

有关向不同主题发送不同类型的另一种技术，见 使用 RoutingKafkaTemplate。

按类型

2.8 版本引入了 DelegatingByTypeSerializer。

@Bean
public ProducerFactory<Integer, Object> producerFactory(Map<String, Object> config) {
    return new DefaultKafkaProducerFactory<>(config,
            null, new DelegatingByTypeSerializer(Map.of(
                    byte[].class, new ByteArraySerializer(),
                    Bytes.class, new BytesSerializer(),
                    String.class, new StringSerializer())));
}

从 2.8.3 版本开始，你可以配置序列化器检查映射键是否可从目标对象赋值，当委托序列化器可以序列化子类时这会很有用。在这种情况下，如果存在歧义匹配，应提供一个有序 Map，例如 LinkedHashMap。

按主题

从 2.8 版本开始，DelegatingByTopicSerializer 和 DelegatingByTopicDeserializer 允许根据主题名称选择序列化器/反序列化器。使用正则 Pattern 查找要使用的实例。该映射可以使用构造函数配置，或通过属性配置（以逗号分隔的 pattern:serializer 列表）。

producerConfigs.put(DelegatingByTopicSerializer.VALUE_SERIALIZATION_TOPIC_CONFIG,
            "topic[0-4]:" + ByteArraySerializer.class.getName()
        + ", topic[5-9]:" + StringSerializer.class.getName());
...
consumerConfigs.put(DelegatingByTopicDeserializer.VALUE_SERIALIZATION_TOPIC_CONFIG,
            "topic[0-4]:" + ByteArrayDeserializer.class.getName()
        + ", topic[5-9]:" + StringDeserializer.class.getName());

当将此用于键时，使用 KEY_SERIALIZATION_TOPIC_CONFIG。

@Bean
public ProducerFactory<Integer, Object> producerFactory(Map<String, Object> config) {
    return new DefaultKafkaProducerFactory<>(config,
            new IntegerSerializer(),
            new DelegatingByTopicSerializer(Map.of(
                    Pattern.compile("topic[0-4]"), new ByteArraySerializer(),
                    Pattern.compile("topic[5-9]"), new StringSerializer())),
                    new JacksonJsonSerializer<Object>());  // default
}

当没有模式匹配时，你可以使用 DelegatingByTopicSerialization.KEY_SERIALIZATION_TOPIC_DEFAULT 和 DelegatingByTopicSerialization.VALUE_SERIALIZATION_TOPIC_DEFAULT 指定要使用的默认序列化器/反序列化器。

额外属性 DelegatingByTopicSerialization.CASE_SENSITIVE（默认 true）设为 false 时，使主题查找不区分大小写。

重试反序列化器

RetryingDeserializer 使用一个委托 Deserializer 和 RetryTemplate，在委托可能因瞬时错误（例如网络问题）导致反序列化失败时重试反序列化。

ConsumerFactory cf = new DefaultKafkaConsumerFactory(myConsumerConfigs,
    new RetryingDeserializer(myUnreliableKeyDeserializer, retryTemplate),
    new RetryingDeserializer(myUnreliableValueDeserializer, retryTemplate));

可以在 RetryingDeserializer 上设置一个恢复回调，以在所有重试用尽时返回一个回退对象。

有关 RetryTemplate 带重试策略、退避等的配置，请参阅 Spring Framework 项目。

Spring Messaging 消息转换

尽管从底层 Kafka Consumer 和 Producer 的角度看，Serializer 和 Deserializer API 相当简单灵活，但在使用 @KafkaListener 或 Spring Integration 的 Apache Kafka 支持 时，你可能在 Spring Messaging 层面需要更多灵活性。为让你能方便地在 org.springframework.messaging.Message 之间转换，Spring for Apache Kafka 提供了一个 MessageConverter 抽象及其 MessagingMessageConverter 实现和它的 JacksonJsonMessageConverter（及子类）定制。你可以把 MessageConverter 直接注入 KafkaTemplate 实例，并通过 AbstractKafkaListenerContainerFactory bean 定义用于 @KafkaListener.containerFactory() 属性。以下示例展示了如何做到这一点：

@Bean
public KafkaListenerContainerFactory<?> kafkaJsonListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setRecordMessageConverter(new JacksonJsonMessageConverter());
    return factory;
}
...
@KafkaListener(topics = "jsonData",
                containerFactory = "kafkaJsonListenerContainerFactory")
public void jsonListener(Cat cat) {
...
}

使用 Spring Boot 时，只需把转换器定义为一个 @Bean，Spring Boot 自动配置就会把它装配进自动配置的模板和容器工厂。

当你使用 @KafkaListener 时，参数类型被提供给消息转换器以协助转换。

备注

这种类型推断只有在 @KafkaListener 注解声明在方法级时才能实现。对于类级 @KafkaListener，载荷类型用于选择要调用哪个 @KafkaHandler 方法，因此在方法被选定之前它必须已经完成转换。

备注

在消费者侧，你可以配置一个 JacksonJsonMessageConverter；它可以处理 byte[]、Bytes 和 String 类型的 ConsumerRecord 值，因此应与 ByteArrayDeserializer、BytesDeserializer 或 StringDeserializer 结合使用。（byte[] 和 Bytes 更高效，因为它们避免了不必要的 byte[] 到 String 的转换。）如果你愿意，也可以配置与该反序列化器对应的 JacksonJsonMessageConverter 的具体子类。

在生产者侧，当你使用 Spring Integration 或 KafkaTemplate.send(Message<?> message) 方法（见 使用 KafkaTemplate）时，你必须配置一个与所配置 Kafka Serializer 兼容的消息转换器。

• StringJacksonJsonMessageConverter 配合 StringSerializer
• BytesJacksonJsonMessageConverter 配合 BytesSerializer
• ByteArrayJacksonJsonMessageConverter 配合 ByteArraySerializer

同样，使用 byte[] 或 Bytes 更高效，因为它们避免了 String 到 byte[] 的转换。

为方便起见，自 2.3 版本起，框架还提供了一个 StringOrBytesSerializer，它可以序列化所有三种值类型，因此可与任意消息转换器配合使用。

从 2.7.1 版本开始，消息载荷转换可以委托给一个 spring-messaging SmartMessageConverter；这使得转换可以基于 MessageHeaders.CONTENT_TYPE 头，例如。

important

KafkaMessageConverter.fromMessage() 方法在出站转换为 ProducerRecord 时被调用，消息载荷位于 ProducerRecord.value() 属性中。KafkaMessageConverter.toMessage() 方法在从 ConsumerRecord 入站转换时被调用，载荷为 ConsumerRecord.value() 属性。SmartMessageConverter.toMessage() 方法被调用以从传给 fromMessage() 的 Message（通常由 KafkaTemplate.send(Message<?> msg)）创建一个新的出站 Message<?>。类似地，在 KafkaMessageConverter.toMessage() 方法中，转换器从 ConsumerRecord 创建新的 Message<?> 之后，SmartMessageConverter.fromMessage() 方法被调用，然后用新转换的载荷创建最终的入站消息。在任一情况下，如果 SmartMessageConverter 返回 null，则使用原始消息。

当在 KafkaTemplate 和监听容器工厂中使用默认转换器时，你通过在模板上调用 setMessagingConverter() 以及通过 @KafkaListener 方法上的 contentTypeConverter 属性来配置 SmartMessageConverter。

示例：

template.setMessagingConverter(mySmartConverter);

@KafkaListener(id = "withSmartConverter", topics = "someTopic",
    contentTypeConverter = "mySmartConverter")
public void smart(Thing thing) {
    ...
}

使用 Spring Data 投影接口

从 2.1.1 版本开始，你可以将 JSON 转换为 Spring Data 投影接口，而不是具体类型。这允许对数据进行高度选择性且低耦合的绑定，包括从 JSON 文档内部多个位置查找值。例如，以下接口可以定义为消息载荷类型：

interface SomeSample {

  @JsonPath({ "$.username", "$.user.name" })
  String getUsername();

}

@KafkaListener(id="projection.listener", topics = "projection")
public void projection(SomeSample in) {
    String username = in.getUsername();
    ...
}

默认情况下，访问器方法将被用于在所接收的 JSON 文档中查找作为字段的属性名。@JsonPath 表达式允许定制值查找，甚至可以定义多个 JSON Path 表达式，从多个位置查找值，直到某个表达式返回实际值。

要启用此特性，使用一个 JacksonProjectingMessageConverter，配置一个适当的委托转换器（用于出站转换和转换非投影接口）。你还必须将 spring-data:spring-data-commons 和 com.jayway.jsonpath:json-path 添加到 classpath。

当用作 @KafkaListener 方法的参数时，该接口类型会像平常一样自动传给转换器。

使用 ErrorHandlingDeserializer

当一个反序列化器无法反序列化消息时，Spring 无法处理该问题，因为它发生在 poll() 返回之前。为解决此问题，引入了 ErrorHandlingDeserializer。该反序列化器委托给一个真正的反序列化器（键或值）。如果委托方未能反序列化记录内容，ErrorHandlingDeserializer 返回一个 null 值，并在头中返回一个包含原因和原始字节的 DeserializationException。当你使用记录级 MessageListener 时，如果 ConsumerRecord 的键或值含有 DeserializationException 头，容器的 ErrorHandler 会以失败的 ConsumerRecord 被调用。该记录不会传给监听器。

或者，你可以通过提供一个 failedDeserializationFunction 来配置 ErrorHandlingDeserializer 创建自定义值，它是一个 Function<FailedDeserializationInfo, T>。调用此函数创建一个 T 的实例，并以通常方式传给监听器。向该函数提供一个 FailedDeserializationInfo 类型的对象，它包含所有上下文信息。你可以在头中找到 DeserializationException（作为序列化的 Java 对象）。有关 ErrorHandlingDeserializer 的更多信息，请参阅 Javadoc。

你可以使用接收键和值 Deserializer 对象的 DefaultKafkaConsumerFactory 构造函数，并装配已用适当委托方配置好的 ErrorHandlingDeserializer 实例。或者，你可以使用消费者配置属性（由 ErrorHandlingDeserializer 使用）来实例化委托方。属性名是 ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS 和 ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS。属性值可以是类或类名。以下示例展示了如何设置这些属性：

... // other props
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, JsonDeserializer.class);
props.put(JsonDeserializer.KEY_DEFAULT_TYPE, "com.example.MyKey")
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.MyValue")
props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example")
return new DefaultKafkaConsumerFactory<>(props);

以下示例使用 failedDeserializationFunction。

public class BadThing extends Thing {

  private final FailedDeserializationInfo failedDeserializationInfo;

  public BadThing(FailedDeserializationInfo failedDeserializationInfo) {
    this.failedDeserializationInfo = failedDeserializationInfo;
  }

  public FailedDeserializationInfo getFailedDeserializationInfo() {
    return this.failedDeserializationInfo;
  }

}

public class FailedThingProvider implements Function<FailedDeserializationInfo, Thing> {

  @Override
  public Thing apply(FailedDeserializationInfo info) {
    return new BadThing(info);
  }

}

前面的示例使用以下配置：

...
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
consumerProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
consumerProps.put(ErrorHandlingDeserializer.VALUE_FUNCTION, FailedThingProvider.class);
...

important

如果消费者配置了 ErrorHandlingDeserializer，重要的是用一个既能处理普通对象、又能处理由反序列化异常产生的原始 byte[] 值的序列化器来配置 KafkaTemplate 及其生产者。模板的泛型值类型应为 Object。一种技术是使用 DelegatingByTypeSerializer；示例如下：

@Bean
public ProducerFactory<String, Object> producerFactory() {
  return new DefaultKafkaProducerFactory<>(producerConfiguration(), new StringSerializer(),
    new DelegatingByTypeSerializer(Map.of(byte[].class, new ByteArraySerializer(),
          MyNormalObject.class, new JacksonJsonSerializer<Object>())));
}

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

当将 ErrorHandlingDeserializer 与批量监听器一起使用时，你必须在消息头中检查反序列化异常。当与 DefaultBatchErrorHandler 一起使用时，你可以使用该头来确定异常发生在哪条记录上，并通过 BatchListenerFailedException 告知错误处理器。

@KafkaListener(id = "test", topics = "test")
void listen(List<Thing> in, @Header(KafkaHeaders.BATCH_CONVERTED_HEADERS) List<Map<String, Object>> headers) {
    for (int i = 0; i < in.size(); i++) {
        Thing thing = in.get(i);
        if (thing == null
                && headers.get(i).get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER) != null) {
            try {
                DeserializationException deserEx = SerializationUtils.byteArrayToDeserializationException(this.logger,
                        headers.get(i).get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER));
                if (deserEx != null) {
                    logger.error(deserEx, "Record at index " + i + " could not be deserialized");
                }
            }
            catch (Exception ex) {
                logger.error(ex, "Record at index " + i + " could not be deserialized");
            }
            throw new BatchListenerFailedException("Deserialization", deserEx, i);
        }
        process(thing);
    }
}

SerializationUtils.byteArrayToDeserializationException() 可用于将该头转换为 DeserializationException。

当消费 List<ConsumerRecord<?, ?>> 时，改用 SerializationUtils.getExceptionFromHeader()：

@KafkaListener(id = "kgh2036", topics = "kgh2036")
void listen(List<ConsumerRecord<String, Thing>> in) {
    for (int i = 0; i < in.size(); i++) {
        ConsumerRecord<String, Thing> rec = in.get(i);
        if (rec.value() == null) {
            DeserializationException deserEx = SerializationUtils.getExceptionFromHeader(rec,
                    SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER, this.logger);
            if (deserEx != null) {
                logger.error(deserEx, "Record at offset " + rec.offset() + " could not be deserialized");
                throw new BatchListenerFailedException("Deserialization", deserEx, i);
            }
        }
        process(rec.value());
    }
}

important

如果你还使用 DeadLetterPublishingRecoverer，为 DeserializationException 发布的记录其 record.value() 将是 byte[] 类型；这不应被序列化。可考虑使用一个 DelegatingByTypeSerializer，配置为对 byte[] 使用 ByteArraySerializer，对所有其他类型使用普通序列化器（Json、Avro 等）。

从 3.1 版本开始，你可以向 ErrorHandlingDeserializer 添加一个 Validator。如果委托 Deserializer 成功反序列化了对象，但该对象未通过校验，则抛出类似于反序列化异常的异常。这允许把原始数据传给错误处理器。当你自己创建反序列化器时，只需调用 setValidator；如果你使用属性配置序列化器，将消费者配置属性 ErrorHandlingDeserializer.VALIDATOR_CLASS 设为你 Validator 的类或全限定类名。使用 Spring Boot 时，此属性名为 spring.kafka.consumer.properties.spring.deserializer.validator.class。

批量监听器的载荷转换

当你使用批量监听器容器工厂时，你还可以在 BatchMessagingMessageConverter 内使用 JacksonJsonMessageConverter 来转换批量消息。更多信息请参阅 序列化、反序列化与消息转换 和 Spring Messaging 消息转换。

默认情况下，转换的类型从监听器参数推断。如果你用一个 DefaultJackson2TypeMapper 配置 JacksonJsonMessageConverter，并将其 TypePrecedence 设为 TYPE_ID（而非默认的 INFERRED），则转换器改用头中的类型信息（如果存在）。例如，这允许监听器方法用接口而非具体类声明。此外，类型转换器支持映射，因此反序列化的目标类型可以与源类型不同（只要数据兼容）。当你使用 类级 @KafkaListener 实例 时这也很有用，因为在确定要调用哪个方法之前，载荷必须已经完成转换。以下示例创建使用此方法的 bean：

@Bean
public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
    factory.setBatchMessageConverter(new BatchMessagingMessageConverter(converter()));
    return factory;
}

@Bean
public JacksonJsonMessageConverter converter() {
    return new JacksonJsonMessageConverter();
}

请注意，要使其生效，转换目标的方法签名必须是一个带有单个泛型参数类型的容器对象，例如：

@KafkaListener(topics = "blc1")
public void listen(List<Foo> foos, @Header(KafkaHeaders.OFFSET) List<Long> offsets) {
    ...
}

请注意，你仍然可以访问批量头。

如果批量转换器有一个支持该特性的记录转换器，你还可以接收一个消息列表，其中载荷按泛型类型转换。以下示例展示了如何做到这一点：

@KafkaListener(topics = "blc3", groupId = "blc3")
public void listen(List<Message<Foo>> fooMessages) {
    ...
}

如果批中某条记录无法转换，其载荷被设为 null 放入目标 payloads 列表。转换异常对该记录以警告级别记录，并作为 List<ConversionException> 的一项存入 KafkaHeaders.CONVERSION_FAILURES 头。目标 @KafkaListener 方法可以执行 Java Stream API 从载荷列表中过滤掉那些 null 值，或对转换异常头做些处理：

@KafkaListener(id = "foo", topics = "foo", autoStartup = "false")
public void listen(List<Foo> list,
         @Header(KafkaHeaders.CONVERSION_FAILURES) List<ConversionException> conversionFailures) {

    for (int i = 0; i < list.size(); i++) {
        if (conversionFailures.get(i) != null) {
            throw new BatchListenerFailedException("Conversion Failed", conversionFailures.get(i), i);
        }
    }
}

ConversionService 定制

从 2.1.1 版本开始，默认的 org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory 用于解析监听器方法调用参数所使用的 org.springframework.core.convert.ConversionService，会配备所有实现以下任一接口的 bean：

• org.springframework.core.convert.converter.Converter
• org.springframework.core.convert.converter.GenericConverter
• org.springframework.format.Formatter

这让你可以进一步定制监听器反序列化，而无需更改 ConsumerFactory 和 KafkaListenerContainerFactory 的默认配置。

important

通过 KafkaListenerConfigurer bean 在 KafkaListenerEndpointRegistrar 上设置自定义 MessageHandlerMethodFactory 会禁用此特性。

向 @KafkaListener 添加自定义 HandlerMethodArgumentResolver

从 2.4.2 版本开始，你可以添加自己的 HandlerMethodArgumentResolver 并解析自定义方法参数。你只需实现 KafkaListenerConfigurer 并使用 KafkaListenerEndpointRegistrar 类的 setCustomMethodArgumentResolvers() 方法。

@Configuration
class CustomKafkaConfig implements KafkaListenerConfigurer {

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        registrar.setCustomMethodArgumentResolvers(
            new HandlerMethodArgumentResolver() {

                @Override
                public boolean supportsParameter(MethodParameter parameter) {
                    return CustomMethodArgument.class.isAssignableFrom(parameter.getParameterType());
                }

                @Override
                public Object resolveArgument(MethodParameter parameter, Message<?> message) {
                    return new CustomMethodArgument(
                        message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, String.class)
                    );
                }
            }
        );
    }

}

你还可以通过向 KafkaListenerEndpointRegistrar bean 添加自定义 MessageHandlerMethodFactory 来完全替换框架的参数解析。如果你这样做，且你的应用需要处理带 null value() 的墓碑记录（例如来自压缩主题），你应向该工厂添加一个 KafkaNullAwarePayloadArgumentResolver；它必须是最后一个解析器，因为它支持所有类型，并且可以匹配没有 @Payload 注解的参数。如果你使用 DefaultMessageHandlerMethodFactory，将此解析器设为最后一个自定义解析器；该工厂会确保此解析器在标准 PayloadMethodArgumentResolver 之前使用，后者对 KafkaNull 载荷一无所知。

另请参阅 空载荷与墓碑记录的日志压缩。