如何在Spring Cloud Stream Kafka中分离生产者与消费者Broker

已成功实现基于Kafka的Spring Cloud Stream流处理，可消费消息后发布至目标Topic，当前功能运行正常，但希望将生产者与消费者的Broker分离（目前两者使用同一Broker），请问该如何配置？

当前配置代码：

spring:
  cloud:
    config:
    function:
      definition: RiskProcessor1;RiskProcessor2;RiskProcessor3;RiskProcessor4;RiskProcessor5
    stream:
      bindings:
        RiskProcessor1-in-0:
          destination: ******
        RiskProcessor1-out-0:
          destination: *****
        RiskProcessor2-in-0:
          destination: *****
        RiskProcessor2-out-0:
          destination: *****
        RiskProcessor3-in-0:
          destination: *****
        RiskProcessor3-out-0:
          destination: *****
        RiskProcessor4-in-0:
          destination: *****
        RiskProcessor4-out-0:
          destination: *****
        RiskProcessor5-in-0:
          destination: *****
        RiskProcessor5-out-0:
          destination: *****
      kafka:
        streams:
          binder:
            brokers: kaas-int.nam.nsroot.net:9093
            functions:
              RiskProcessor1:
                applicationId: RiskProcessor1_development
              RiskProcessor2:
                applicationId: RiskProcessor2_development
              RiskProcessor3:
                applicationId: RiskProcessor3_development
              RiskProcessor4:
                applicationId: RiskProcessor4_development
              RiskProcessor5:
                applicationId: RiskProcessor5_development
            configuration:
              commit.interval.ms: 1000
              security.protocol: SSL
              default:
                deserialization:
                  exception:
                    handler: org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
              schema:
                registry:
                  url: *****:9081
          default:
            consumer:
              keySerde: ****
              valueSerde: ****
            producer:
              keySerde: ****
              valueSerde: ****

要实现生产者与消费者Broker分离，需通过Spring Cloud Stream的多Binder配置实现，核心是定义两个独立的Kafka Streams Binder，分别对应消费者（输入）和生产者（输出）的Broker，再给每个绑定指定对应的Binder。

配置修改步骤 #

1. 新增binders节点，定义两个Binder：kafka-in（消费者用）和kafka-out（生产者用），各自配置对应的Broker地址、安全参数、Schema Registry等。
2. 给所有输入绑定（*in-0）添加binder: kafka-in属性，指定使用消费者Broker的Binder。
3. 给所有输出绑定（*out-0）添加binder: kafka-out属性，指定使用生产者Broker的Binder。

修改后的完整配置示例 #

spring:
  cloud:
    function:
      definition: RiskProcessor1;RiskProcessor2;RiskProcessor3;RiskProcessor4;RiskProcessor5
    stream:
      bindings:
        # 输入绑定关联消费者Binder
        RiskProcessor1-in-0:
          destination: ******
          binder: kafka-in
        # 输出绑定关联生产者Binder
        RiskProcessor1-out-0:
          destination: *****
          binder: kafka-out
        RiskProcessor2-in-0:
          destination: *****
          binder: kafka-in
        RiskProcessor2-out-0:
          destination: *****
          binder: kafka-out
        RiskProcessor3-in-0:
          destination: *****
          binder: kafka-in
        RiskProcessor3-out-0:
          destination: *****
          binder: kafka-out
        RiskProcessor4-in-0:
          destination: *****
          binder: kafka-in
        RiskProcessor4-out-0:
          destination: *****
          binder: kafka-out
        RiskProcessor5-in-0:
          destination: *****
          binder: kafka-in
        RiskProcessor5-out-0:
          destination: *****
          binder: kafka-out
      kafka:
        streams:
          binders:
            # 消费者Binder配置（原Broker配置迁移至此）
            kafka-in:
              type: kafka-streams
              environment:
                spring:
                  cloud:
                    stream:
                      kafka:
                        streams:
                          binder:
                            brokers: 消费者Broker地址:9093
                            functions:
                              RiskProcessor1:
                                applicationId: RiskProcessor1_development
                              RiskProcessor2:
                                applicationId: RiskProcessor2_development
                              RiskProcessor3:
                                applicationId: RiskProcessor3_development
                              RiskProcessor4:
                                applicationId: RiskProcessor4_development
                              RiskProcessor5:
                                applicationId: RiskProcessor5_development
                            configuration:
                              commit.interval.ms: 1000
                              security.protocol: SSL
                              default:
                                deserialization:
                                  exception:
                                    handler: org.apache.kafka.streams.errors.LogAndContinueExceptionHandler
                              schema:
                                registry:
                                  url: 消费者侧Schema Registry地址:9081
            # 生产者Binder配置（替换为目标Broker参数）
            kafka-out:
              type: kafka-streams
              environment:
                spring:
                  cloud:
                    stream:
                      kafka:
                        streams:
                          binder:
                            brokers: 生产者Broker地址:9093
                            configuration:
                              security.protocol: SSL
                              schema:
                                registry:
                                  url: 生产者侧Schema Registry地址:9081
          default:
            consumer:
              keySerde: ****
              valueSerde: ****
            producer:
              keySerde: ****
              valueSerde: ****

关键说明 #

• 两个Binder的配置相互独立，可以根据实际需求调整各自的security.protocol、schema.registry.url等参数。
• Kafka Streams的applicationId只需在消费者Binder中配置，因为流处理的元数据（如状态存储）与输入Broker关联。
• 如果两个Binder有通用配置（如Serde），可以保留在default节点，无需重复配置。

内容的提问来源于stack exchange，提问作者Himanshu Mahajan