记录序列化和反序列化

Kafka Streams 绑定器允许您以两种方式序列化和反序列化记录。一种是 Kafka 提供的原生序列化和反序列化功能，另一种是 Spring Cloud Stream 框架的消息转换功能。让我们看看一些细节。

入站反序列化

键始终使用原生 Serdes 反序列化。

对于值，默认情况下，入站的反序列化由 Kafka 本地执行。请注意，这是 Kafka Streams 绑定器先前版本默认行为的重大更改，在先前版本中，反序列化由框架完成。

Kafka Streams 绑定器将尝试通过查看 `java.util.function.Function|Consumer` 的类型签名来推断匹配的 `Serde` 类型。以下是它匹配 Serdes 的顺序。

  • 如果应用程序提供类型为 `Serde` 的 bean，并且如果返回类型使用传入键或值类型的实际类型进行参数化，那么它将使用该 `Serde` 进行入站反序列化。例如，如果您在应用程序中具有以下内容，则绑定器检测到传入 `KStream` 的值类型与 `Serde` bean 上参数化的类型匹配。它将使用它进行入站反序列化。

@Bean
public Serde<Foo> customSerde() {
 ...
}

@Bean
public Function<KStream<String, Foo>, KStream<String, Foo>> process() {
}

  • 接下来，它查看类型，并查看它们是否为 Kafka Streams 公开的类型之一。如果是，则使用它们。以下是绑定器将尝试从 Kafka Streams 匹配的 Serde 类型。

    Integer, Long, Short, Double, Float, byte[], UUID and String.

  • 如果 Kafka Streams 提供的 Serdes 都不匹配类型，则它将使用 Spring Kafka 提供的 JsonSerde。在这种情况下，绑定器假设类型是 JSON 友好的。如果您有多个值对象作为输入，这将很有用，因为绑定器将在内部将它们推断为正确的 Java 类型。不过，在回退到 JsonSerde 之前，绑定器会检查 Kafka Streams 配置中设置的默认 Serde，以查看它是否是一个可以与传入 KStream 类型匹配的 Serde。

如果以上策略均无效，则应用程序必须通过配置提供 Serde。这可以通过两种方式配置 - 绑定或默认。

首先，绑定器将查看绑定级别是否提供了 Serde。例如，如果您有以下处理器，

@Bean
public BiFunction<KStream<CustomKey, AvroIn1>, KTable<CustomKey, AvroIn2>, KStream<CustomKey, AvroOutput>> process() {...}

那么，您可以使用以下方法提供绑定级别的 Serde

spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.keySerde=CustomKeySerde
spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.keySerde=CustomKeySerde
spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
如果您按输入绑定提供上述 Serde，那么它将具有更高的优先级，绑定器将避免任何 Serde 推断。

如果您希望将默认键/值 Serdes 用于入站反序列化，您可以在绑定器级别执行此操作。

spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde

如果您不希望使用 Kafka 提供的本机解码，您可以依赖 Spring Cloud Stream 提供的消息转换功能。由于本机解码是默认设置，为了让 Spring Cloud Stream 反序列化入站值对象，您需要显式禁用本机解码。

例如，如果您有与上面相同的 BiFunction 处理器，那么 spring.cloud.stream.bindings.process-in-0.consumer.nativeDecoding: false 您需要为所有输入分别禁用本机解码。否则，本机解码仍将应用于您未禁用的那些输入。

默认情况下，Spring Cloud Stream 将使用 application/json 作为内容类型并使用适当的 json 消息转换器。您可以使用以下属性和适当的 MessageConverter bean 使用自定义消息转换器。

spring.cloud.stream.bindings.process-in-0.contentType

出站序列化

出站序列化与入站反序列化遵循相同的规则。与入站反序列化一样，Spring Cloud Stream 前几个版本的一个主要变化是，出站的序列化由 Kafka 本地处理。在绑定器的 3.0 版本之前，这是由框架本身完成的。

出站的键始终由 Kafka 使用匹配的 Serde 序列化，该 Serde 由绑定器推断。如果它无法推断键的类型，则需要使用配置指定。

值序列化器/反序列化器是使用与入站反序列化相同的规则推断的。首先，它会匹配以查看出站类型是否来自应用程序中提供的 bean。如果不是，它会检查它是否与 Kafka 公开的 Serde 匹配，例如 - Integer、Long、Short、Double、Float、byte[]、UUID 和 String。如果这不起作用，那么它将回退到 Spring Kafka 项目提供的 JsonSerde，但首先查看默认的 Serde 配置以查看是否有匹配项。请记住，所有这些都对应用程序是透明的。如果这些都不起作用，那么用户必须通过配置提供要使用的 Serde。

假设您正在使用与上面相同的 BiFunction 处理器。然后您可以配置出站键/值 Serdes 如下。

spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.keySerde=CustomKeySerde
spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde

如果 Serde 推断失败，并且没有提供绑定级别的 Serdes，那么绑定器将回退到 JsonSerde，但会查看默认的 Serdes 以进行匹配。

默认的 serdes 的配置方式与上面描述的反序列化相同。

spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde

如果您的应用程序使用分支功能并具有多个输出绑定，那么这些必须针对每个绑定进行配置。再次，如果绑定器能够推断 Serde 类型，则您无需进行此配置。

如果您不希望使用 Kafka 提供的本机编码，而是希望使用框架提供的消息转换，那么您需要显式禁用本机编码，因为本机编码是默认设置。例如，如果您有与上面相同的 BiFunction 处理器，那么 spring.cloud.stream.bindings.process-out-0.producer.nativeEncoding: false 您需要在分支的情况下为所有输出单独禁用本机编码。否则，本机编码将仍然应用于您未禁用的那些。

当 Spring Cloud Stream 完成转换时，默认情况下，它将使用 application/json 作为内容类型，并使用适当的 json 消息转换器。您可以使用以下属性和相应的 MessageConverter bean 来使用自定义消息转换器。

spring.cloud.stream.bindings.process-out-0.contentType

当禁用本机编码/解码时，绑定器不会像本机 Serdes 那样进行任何推断。应用程序需要显式提供所有配置选项。因此，通常建议使用默认的序列化/反序列化选项，并在编写 Spring Cloud Stream Kafka Streams 应用程序时坚持使用 Kafka Streams 提供的本机序列化/反序列化。您必须使用框架提供的消息转换功能的一种情况是，当您的上游生产者使用特定的序列化策略时。在这种情况下，您希望使用匹配的反序列化策略，因为本机机制可能会失败。当依赖于默认的 Serde 机制时，应用程序必须确保绑定器具有前进的途径，并使用正确的 Serde 正确映射入站和出站，否则可能会出现故障。

值得一提的是，上面概述的数据反序列化方法仅适用于处理器的边缘，即 - 入站和出站。您的业务逻辑可能仍然需要调用明确需要Serde对象的 Kafka Streams API。这些仍然是应用程序的责任，必须由开发人员相应地处理。