如何配置Spring Cloud Azure Kafka Binding绑定多个Event Hub命名空间？

核心思路 #

使用Spring Cloud Stream原生的*多绑定器（Multi Binder）*能力，为两个不同的Event Hub命名空间分别注册独立的Kafka绑定器实例，分别关联消费通道和生产通道即可实现跨命名空间的消息流转。

1. 依赖引入 #

和单命名空间对接配置的依赖一致，只需保证项目中引入了Spring Cloud Stream Kafka绑定器以及Azure Event Hub Kafka兼容层相关依赖即可，核心依赖坐标如下：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>
<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-eventhubs-kafka</artifactId>
</dependency>

2. 配置文件编写 #

在application.yml中配置两个独立的绑定器，分别对应源（消费）命名空间和目标（生产）命名空间，示例配置如下：

spring:
  cloud:
    stream:
      # 注册两个独立的Kafka绑定器
      binders:
        # 源Event Hub命名空间绑定器，用于消费消息
        source-eventhub-binder:
          type: kafka
          environment:
            spring:
              kafka:
                bootstrap-servers: 源命名空间名称.servicebus.windows.net:9093
                properties:
                  sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="源命名空间的连接字符串";
                  sasl.mechanism: PLAIN
                  security.protocol: SASL_SSL
        # 目标Event Hub命名空间绑定器，用于发送消息
        target-eventhub-binder:
          type: kafka
          environment:
            spring:
              kafka:
                bootstrap-servers: 目标命名空间名称.servicebus.windows.net:9093
                properties:
                  sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="目标命名空间的连接字符串";
                  sasl.mechanism: PLAIN
                  security.protocol: SASL_SSL
      # 通道绑定配置
      bindings:
        # 消费输入通道，绑定源命名空间的Event Hub实例
        transferMessage-in-0:
          destination: 源Event Hub实例名称
          group: 自定义消费者组名称
          binder: source-eventhub-binder
        # 生产输出通道，绑定目标命名空间的Event Hub实例
        transferMessage-out-0:
          destination: 目标Event Hub实例名称
          binder: target-eventhub-binder
      # 声明业务处理函数
      function:
        definition: transferMessage

3. 业务代码实现 #

只需定义一个函数式Bean实现消息中转逻辑即可，如有自定义处理、过滤、转换需求可直接在函数中扩展：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.function.Function;

@Configuration
public class MessageTransferConfig {

    @Bean
    public Function<String, String> transferMessage() {
        return message -> {
            // 此处可添加自定义业务处理逻辑
            return message;
        };
    }
}

注意事项 #

• 源命名空间的连接字符串需要具备对应Event Hub实例的Listen权限，目标命名空间的连接字符串需要具备对应Event Hub实例的Send权限
• 如需保留消息头、分区等元数据，可将函数的入参出参改为org.springframework.messaging.Message<String>类型手动处理
• 需确认应用所在网络已开通两个Event Hub命名空间的访问权限，包括防火墙白名单、专用链路等配置

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