跳到主要内容
版本:5.0.1

Kafka Streams 应用程序中的基于事件类型路由

DeepSeek V3 中英对照 Event type based routing in Kafka Streams applications

基于常规消息通道的绑定器中可用的路由功能在 Kafka Streams 绑定器中不受支持。然而,Kafka Streams 绑定器仍然通过入站记录上的事件类型记录头提供路由功能。

要启用基于事件类型的路由,应用程序必须提供以下属性。

spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer.eventTypes。

这可以是一个逗号分隔的值。

例如,假设我们有这个函数:

@Bean
public Function<KStream<Integer, Foo>, KStream<Integer, Foo>> process() {
return input -> input;
}

同时,我们假设只有当传入记录的事件类型为 foo 或 bar 时,才希望执行此函数中的业务逻辑。这可以通过绑定上的 eventTypes 属性来表达,如下所示。

spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypes=foo,bar

现在,当应用程序运行时,绑定器会检查每个传入记录的头部 event_type,并查看其值是否设置为 foo 或 bar。如果未找到其中任何一个,则函数执行将被跳过。

默认情况下,绑定器期望记录头键为 event_type,但这可以按绑定进行更改。例如,如果我们希望将此绑定上的头键更改为 my_event 而非默认值,可以按如下方式更改。

spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypeHeaderKey=my_event.

在使用 Kafka Streams 绑定器的事件路由功能时,系统会使用字节数组 Serde 来反序列化所有传入记录。只有当记录头信息匹配事件类型时,才会使用实际的 Serde 通过配置或推断的 Serde 进行正确的反序列化。如果在绑定上设置了反序列化异常处理器,这会导致问题,因为预期的反序列化仅在堆栈下层发生,从而引发意外错误。为了解决此问题,您可以在绑定上设置以下属性,强制绑定器使用配置或推断的 Serde,而不是字节数组 Serde。

spring.cloud.stream.kafka.streams.bindings.<process-in-0>.consumer.useConfiguredSerdeWhenRoutingEvents

这样一来,应用程序在使用事件路由功能时就能立即检测到反序列化问题,并可以采取适当的处理决策。