spring cloud stream rocketmq 多生产通道 多消费通道

Spring Cloud Stream RocketMQ 支持通过配置多个生产通道和消费通道。在 RocketMQ 中，生产通道被称为生产者组，消费通道被称为消费者组。可以通过在应用程序的配置文件中指定多个生产者组和消费者组来配置多个通道。

以下是一个示例配置，其中定义了两个生产者组和两个消费者组：

spring:
  cloud:
    stream:
      rocketmq:
        binder:
          brokers: ${rocketmq.broker-addresses}
        bindings:
          output1:
            producer:
              group: my-producer-group-1
          output2:
            producer:
              group: my-producer-group-2
          input1:
            consumer:
              group: my-consumer-group-1
          input2:
            consumer:
              group: my-consumer-group-2

在这个配置中，我们定义了两个生产者组（`my-producer-group-1` 和 `my-producer-group-2`）和两个消费者组（`my-consumer-group-1` 和 `my-consumer-group-2`）。我们还指定了每个生产者组和消费者组对应的输出和输入通道。在这个例子中，我们有两个输出通道（`output1` 和 `output2`）和两个输入通道（`input1` 和 `input2`）。

在应用程序中，我们可以使用 `@StreamListener` 注解来监听每个输入通道，然后处理接收到的消息。我们还可以使用 `@Output` 注解来指定要发送消息的输出通道。

例如，假设我们有一个名为 `MyMessage` 的 POJO 类，我们可以使用以下代码将消息发送到 `output1`：

@Service
@EnableBinding(Source.class)
public class MyProducer {
 
    @Autowired
    private MessageChannel output1;
 
    public void sendMessage(MyMessage message) {
        output1.send(MessageBuilder.withPayload(message).build());
    }
 
}

然后，我们可以使用以下代码监听 `input1` 并处理接收到的消息：

@Service
@EnableBinding(Sink.class)
public class MyConsumer {
 
    @StreamListener(Sink.INPUT)
    public void handleMessage(MyMessage message) {
        // 处理消息
    }
 
}

类似地，我们可以使用 `output2` 和 `input2` 来定义第二个通道。