输出与输入绑定目标相同时Stream Bridge测试失败问题排查

问题描述 #

在使用Spring Cloud Stream的Stream Bridge时，当输出绑定目标与输入绑定目标相同时，测试无法正常获取消息。

配置信息 #

bindings:
  receiveEvents-in-0:
    destination: internal
    group: a-certain-group
  sendMessages-out-0:
    destination: internal
rabbit:
  bindings:
    receiveEvents-in-0:
      consumer:
        bindingRoutingKey: 'bla'
    sendMessages-out-0:
      producer:
        routingKeyExpression: headers['routing-key']

测试代码 #

@SpringBootTest
@Import(TestChannelBinderConfiguration.class)
class StreamBridgeIntegrationTest {

private static final int TIMEOUT = 1000;

@Autowired
private OutputDestination output;

@Test
void test() throws IOException {
    var message = // 带有routing key头的特定消息
    streamBridge.send("internal", message);
    var sentMessage = output.receive(TIMEOUT, "internal").getPayload(); // 返回null
    // 断言逻辑
}

已尝试的操作 #

• 使用绑定名称发送消息，无效：

  streamBridge.send("sendMessages-out-0", message); // 无效
  var sentMessage = output.receive(TIMEOUT, "sendMessages-out-0").getPayload(); // 仍返回null
  

• 修改输出绑定目标为random，测试通过，但不符合需求：

  bindings:
      receiveEvents-in-0:
        destination: internal
        group: a-certain-group
      sendMessages-out-0:
        destination: random
  

  streamBridge.send("random", message); // 有效
  var sentMessage = output.receive(TIMEOUT, "random").getPayload(); // 成功获取消息
  

• 注释输入绑定配置，测试通过，但不符合需求：

  bindings:
  #    receiveEvents-in-0:
  #      destination: internal
  #      group: a-certain-group
      sendMessages-out-0:
        destination: internal
  

  streamBridge.send("internal", message); // 有效
  var sentMessage = output.receive(TIMEOUT, "internal").getPayload(); // 成功获取消息
  

需求 #

需要将消息发送到名为internal的RabbitMQ主题交换机，同时从同一目标接收消息（指定路由键），且消费者仅接收集群外部发来的消息，不接收自身发布的消息。

解决方案 #

1. 调整测试逻辑，改用InputDestination接收消息 #

测试环境中，当输入输出目标相同时，OutputDestination无法捕获消息的原因是：测试binder会直接将生产者发送的消息路由到对应输入绑定的消费者通道，而非留存到OutputDestination。此时应改用InputDestination来获取消费者端的消息，修改测试代码如下：

@SpringBootTest
@Import(TestChannelBinderConfiguration.class)
class StreamBridgeIntegrationTest {

private static final int TIMEOUT = 1000;

@Autowired
private StreamBridge streamBridge;

@Autowired
private InputDestination input;

@Test
void test() throws IOException {
    var message = MessageBuilder.withPayload("test")
            .setHeader("routing-key", "bla") // 匹配消费者的bindingRoutingKey
            .build();
    streamBridge.send("internal", message);
    var receivedMessage = input.receive(TIMEOUT, "receiveEvents-in-0").getPayload();
    // 执行断言逻辑
}

2. 生产环境配置：避免消费者自消费 #

要让消费者不接收自身发布的消息，可通过Spring Cloud Stream的consumer.excludeSelf属性实现（RabbitMQ binder原生支持），修改配置如下：

bindings:
  receiveEvents-in-0:
    destination: internal
    group: a-certain-group
    consumer:
      excludeSelf: true # 开启自消费过滤
  sendMessages-out-0:
    destination: internal
rabbit:
  bindings:
    receiveEvents-in-0:
      consumer:
        bindingRoutingKey: 'bla'
    sendMessages-out-0:
      producer:
        routingKeyExpression: headers['routing-key']

该配置会让消费者自动过滤掉当前实例发布的消息，仅接收外部实例发送的消息。

3. 关于同一目标复用的逻辑说明 #

同一目标同时作为输入输出本身不存在逻辑问题，这是RabbitMQ主题交换机的典型使用场景——生产者和消费者共用同一个交换机，通过路由键区分消息流向。问题仅出在测试binder的内部流转逻辑，生产环境中该模式可以正常运行。

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