带有基于 Kafka Streams 绑定器和常规 Kafka 绑定器的多绑定器
你可以有一个应用程序,其中包含基于常规 Kafka 绑定器的函数/消费者/生产者,以及一个基于 Kafka Streams 的处理器。但是,你不能在单个函数或消费者中混合使用它们。
以下是一个示例,其中你在同一个应用程序中拥有两个基于绑定器的组件。
@Bean
public Function<String, String> process() {
return s -> s;
}
@Bean
public Function<KStream<Object, String>, KStream<?, WordCount>> kstreamProcess() {
return input -> input;
}
这是配置中的相关部分
spring.cloud.function.definition=process;kstreamProcess
spring.cloud.stream.bindings.process-in-0.destination=foo
spring.cloud.stream.bindings.process-out-0.destination=bar
spring.cloud.stream.bindings.kstreamProcess-in-0.destination=bar
spring.cloud.stream.bindings.kstreamProcess-out-0.destination=foobar
如果你有上述相同的应用程序,但它处理两个不同的 Kafka 集群,例如,常规的 `process` 在 Kafka 集群 1 和集群 2 上都进行操作(从集群 1 接收数据并发送到集群 2),而 Kafka Streams 处理器在 Kafka 集群 2 上进行操作。那么你必须使用 Spring Cloud Stream 提供的多绑定器功能。
在这种情况下,你的配置可能会如下改变。
# multi binder configuration
spring.cloud.stream.binders.kafka1.type: kafka
spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-1} #Replace kafkaCluster-1 with the approprate IP of the cluster
spring.cloud.stream.binders.kafka2.type: kafka
spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} #Replace kafkaCluster-2 with the approprate IP of the cluster
spring.cloud.stream.binders.kafka3.type: kstream
spring.cloud.stream.binders.kafka3.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2} #Replace kafkaCluster-2 with the approprate IP of the cluster
spring.cloud.function.definition=process;kstreamProcess
# From cluster 1 to cluster 2 with regular process function
spring.cloud.stream.bindings.process-in-0.destination=foo
spring.cloud.stream.bindings.process-in-0.binder=kafka1 # source from cluster 1
spring.cloud.stream.bindings.process-out-0.destination=bar
spring.cloud.stream.bindings.process-out-0.binder=kafka2 # send to cluster 2
# Kafka Streams processor on cluster 2
spring.cloud.stream.bindings.kstreamProcess-in-0.destination=bar
spring.cloud.stream.bindings.kstreamProcess-in-0.binder=kafka3
spring.cloud.stream.bindings.kstreamProcess-out-0.destination=foobar
spring.cloud.stream.bindings.kstreamProcess-out-0.binder=kafka3
请注意以上配置。我们有两种绑定器,但总共有 3 个绑定器,第一个是基于集群 1 的常规 Kafka 绑定器 (`kafka1`),然后是基于集群 2 的另一个 Kafka 绑定器 (`kafka2`),最后是 `kstream` 绑定器 (`kafka3`)。应用程序中的第一个处理器从 `kafka1` 接收数据并发布到 `kafka2`,其中两个绑定器都基于常规 Kafka 绑定器但属于不同集群。第二个处理器是 Kafka Streams 处理器,它从 `kafka3` 消费数据,`kafka3` 与 `kafka2` 是同一个集群,但绑定器类型不同。
由于 Kafka Streams 绑定器系列中有三种不同的绑定器类型——`kstream`、`ktable` 和 `globalktable`——如果你的应用程序有多个基于这些绑定器的绑定,则需要明确将其作为绑定器类型提供。
例如,如果你的处理器如下所示,
@Bean
public Function<KStream<Long, Order>,
Function<KTable<Long, Customer>,
Function<GlobalKTable<Long, Product>, KStream<Long, EnrichedOrder>>>> enrichOrder() {
...
}
那么,这必须在多绑定器场景中配置如下。请注意,这仅在你有真正的多绑定器场景时才需要,即单个应用程序中有多个处理器处理多个集群。在这种情况下,绑定器需要明确提供绑定,以区分其他处理器的绑定器类型和集群。
spring.cloud.stream.binders.kafka1.type: kstream
spring.cloud.stream.binders.kafka1.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2}
spring.cloud.stream.binders.kafka2.type: ktable
spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2}
spring.cloud.stream.binders.kafka3.type: globalktable
spring.cloud.stream.binders.kafka3.environment.spring.cloud.stream.kafka.streams.binder.brokers=${kafkaCluster-2}
spring.cloud.stream.bindings.enrichOrder-in-0.binder=kafka1 #kstream
spring.cloud.stream.bindings.enrichOrder-in-1.binder=kafka2 #ktablr
spring.cloud.stream.bindings.enrichOrder-in-2.binder=kafka3 #globalktable
spring.cloud.stream.bindings.enrichOrder-out-0.binder=kafka1 #kstream
# rest of the configuration is omitted.