如何复用同一Processor实现多组Spring Cloud Stream Kafka Streams绑定？

问题背景 #

使用spring-cloud-stream-binder-kafka-streams 4.0.3版本，存在数量不固定的输入/输出Topic（如topic-in-1/2/3、topic-out-1/2/3），已定义通用处理函数trace：

@Bean
public Function<KStream<String, String>, KStream<String, String>> trace() {
    return input -> input.peek((key, value) -> log.trace("Record processed {}", value));
}

需求是复用该函数构建多组独立流链路（如单输入单输出、多输入单输出的组合），无需每次新增链路就创建重复功能的新函数。

解决方案：通过配置实现函数多实例绑定 #

Spring Cloud Stream支持通过配置为同一个函数创建多个独立绑定实例，每个实例对应一组流处理链路，无需修改代码。核心是利用函数定义的重复声明和绑定后缀区分不同实例。

场景1：多组单输入单输出链路 #

要实现topic-in-1 > trace > topic-out-1、topic-in-2 > trace > topic-out-2、topic-in-3 > trace > topic-out-3三组独立链路，配置如下：

# 声明3个trace函数实例，用分号分隔
spring.cloud.stream.function.definition=trace;trace;trace

# 第一组链路配置
spring.cloud.stream.bindings.trace-in-0.destination=topic-in-1
spring.cloud.stream.bindings.trace-out-0.destination=topic-out-1
spring.cloud.stream.kafka.streams.bindings.trace-in-0.consumer.application-id=trace-stream-1

# 第二组链路配置
spring.cloud.stream.bindings.trace-in-1.destination=topic-in-2
spring.cloud.stream.bindings.trace-out-1.destination=topic-out-2
spring.cloud.stream.kafka.streams.bindings.trace-in-1.consumer.application-id=trace-stream-2

# 第三组链路配置
spring.cloud.stream.bindings.trace-in-2.destination=topic-in-3
spring.cloud.stream.bindings.trace-out-2.destination=topic-out-3
spring.cloud.stream.kafka.streams.bindings.trace-in-2.consumer.application-id=trace-stream-3

• 每个trace实例对应一组链路，通过-in-N/-out-N的后缀区分（N从0开始递增）
• 必须为每个实例指定唯一的application-id，避免Kafka Streams拓扑冲突

场景2：多输入合并+单输入的组合链路 #

要实现topic-in-1、topic-in-2 > trace > topic-out-1、topic-in-3 > trace > topic-out-2两组链路，配置如下：

# 声明2个trace函数实例
spring.cloud.stream.function.definition=trace;trace

# 第一组：多输入合并到单输出
spring.cloud.stream.bindings.trace-in-0.destination=topic-in-1,topic-in-2
spring.cloud.stream.bindings.trace-out-0.destination=topic-out-1
spring.cloud.stream.kafka.streams.bindings.trace-in-0.consumer.application-id=trace-stream-combined

# 第二组：单输入单输出
spring.cloud.stream.bindings.trace-in-1.destination=topic-in-3
spring.cloud.stream.bindings.trace-out-1.destination=topic-out-2
spring.cloud.stream.kafka.streams.bindings.trace-in-1.consumer.application-id=trace-stream-single

• 多输入可通过在destination中用逗号分隔多个Topic实现
• 同样每个实例需要独立的application-id

新增链路的扩展方式 #

后续新增链路时，只需：

1. 在spring.cloud.stream.function.definition中追加;trace（增加一个实例）
2. 新增对应序号的绑定配置（如trace-in-3、trace-out-3）
3. 为新实例指定唯一的application-id

无需修改任何Java代码，完全通过配置扩展。

内容的提问来源于stack exchange，提问作者Sergey Dvoreckih