Spring Cloud Function绑定多RabbitMQ实例单消费函数问题

尝试通过单个Spring消费函数连接两个不同RabbitMQ实例的队列，但消费函数仅绑定application.yml中声明的第一个binder，第二个绑定配置被忽略。

当前application.yml配置：

function:
  definition: processMessage
stream:
  binders:
    rabbit1:
      type: rabbit
      environment:
        spring:
          rabbitmq:
            host: ${host}
            port: ${port}
            username: ${username}
            password: ${password}
            virtual-host: ${virtual-host}
    rabbit2:
      type: rabbit
      environment:
        spring:
          rabbitmq:
            host: ${host}
            port: ${port}
            username: ${username}
            password: ${password}
            virtual-host: ${virtual-host}
  bindings:
    processMessage-in-0:
      destination: exchange1
      group: queue1
      binder: rabbit1
    processMessage-in-1:
      destination: exchange1
      group: queue1
      binder: rabbit2

消费函数代码：

@Bean
public Consumer<Long> processMessage() {
    return (message) -> System.out.println(" Read from processMessage {} " + message.longValue());
}

期望同一函数能读取两个RabbitMQ实例的队列消息，但目前仅能读取processMessage-in-0绑定的实例，不想通过创建两个独立消费函数解决问题，询问是否有可行方案。

1. 调整函数为Flux输入并启用流合并 #

Spring Cloud Function支持将多输入绑定合并为一个流，只需将消费函数的输入类型改为Flux<Long>，并通过配置启用合并逻辑：

修改后的消费函数：

@Bean
public Consumer<Flux<Long>> processMessage() {
    return flux -> flux.subscribe(message -> 
        System.out.println(" Read from processMessage {} " + message.longValue())
    );
}

补充application.yml配置：

spring:
  cloud:
    stream:
      function:
        definition: processMessage
      bindings:
        processMessage-in-0:
          destination: exchange1
          group: queue1
          binder: rabbit1
        processMessage-in-1:
          destination: exchange1
          group: queue1
          binder: rabbit2
      bindings.processMessage-in-0.consumer.multiplex: true
      bindings.processMessage-in-1.consumer.multiplex: true

该配置会将两个不同binder的输入流合并，统一交给同一个消费函数处理。

2. 使用桥接绑定统一消息通道 #

通过桥接函数将两个RabbitMQ实例的消息转发到同一个中间通道，再让目标消费函数监听该通道：

添加桥接函数：

@Bean
public Function<Flux<Long>, Flux<Long>> bridge() {
    return flux -> flux;
}

更新application.yml配置：

spring:
  cloud:
    stream:
      function:
        definition: bridge;processMessage
      bindings:
        # 绑定两个RabbitMQ实例的队列到桥接函数输入
        bridge-in-0:
          destination: exchange1
          group: queue1
          binder: rabbit1
        bridge-in-1:
          destination: exchange1
          group: queue1
          binder: rabbit2
        # 桥接函数输出到统一通道
        bridge-out-0:
          destination: unified-channel
          binder: rabbit1
        # 目标消费函数监听统一通道
        processMessage-in-0:
          destination: unified-channel
          group: unified-queue
          binder: rabbit1

3. 基于Spring Integration共享消费逻辑 #

通过自定义两个输入通道，分别绑定不同的RabbitMQ实例，再将两个通道的消息路由到同一消费逻辑：

自定义通道与集成流：

@Bean
public MessageChannel inputChannel1() {
    return new DirectChannel();
}

@Bean
public MessageChannel inputChannel2() {
    return new DirectChannel();
}

// 共享的消费逻辑
private void handleMessage(Long payload) {
    System.out.println(" Read from processMessage {} " + payload.longValue());
}

@Bean
public IntegrationFlow flow1(MessageChannel inputChannel1) {
    return IntegrationFlows.from(inputChannel1)
        .handle(message -> handleMessage((Long) message.getPayload()))
        .get();
}

@Bean
public IntegrationFlow flow2(MessageChannel inputChannel2) {
    return IntegrationFlows.from(inputChannel2)
        .handle(message -> handleMessage((Long) message.getPayload()))
        .get();
}

更新application.yml配置：

spring:
  cloud:
    stream:
      bindings:
        inputChannel1-in-0:
          destination: exchange1
          group: queue1
          binder: rabbit1
        inputChannel2-in-0:
          destination: exchange1
          group: queue1
          binder: rabbit2

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