Spring Cloud Stream测试绑定器OutputChannel订阅不触发问题排查

核心原因：Spring Cloud Stream输出通道的代理机制 #

你遇到的问题本质是用户自定义的@Output通道Bean和Spring Cloud Stream绑定器实际使用的通道不是同一个对象：

• 当你用@Output("planeEventProducer")定义通道时，Spring Cloud Stream会自动创建一个绑定器代理通道，实际发送消息时，是通过这个代理通道转发到Kafka，而非你自己定义的SubscribableChannel Bean。
• 你直接订阅自己定义的planeEventProducer Bean，相当于监听了一个“空壳”通道，消息根本不会流经这里，所以回调永远不会触发。

为什么调整配置后部分通道生效？ #

你提到修改配置后planeEventProducer-out-0能收到消息，是因为这个名称是Spring Cloud Stream绑定器自动生成的内部实际通道名（格式为<outputBeanName>-out-<index>，单输出场景下index为0），绑定器的消息流会直接经过这个通道，所以订阅它能捕获到消息。而你原来的planeEventProducer只是一个标记性的Bean，没有参与实际的消息转发流程。

验证输出消息的可行方案 #

方案1：订阅绑定器自动生成的内部通道 #

直接针对绑定器生成的通道名（比如planeEventProducer-out-0）进行订阅，就能捕获到发送到Kafka的消息：

@Autowired
@Qualifier("planeEventProducer-out-0")
private SubscribableChannel actualOutputChannel;

// 在测试中订阅该通道
actualOutputChannel.subscribe(message -> {
    // 收集消息用于验证
});

方案2：使用ChannelInterceptor拦截输出 #

自定义一个拦截器，绑定到绑定器的输出通道上，统一收集消息：

@Component
public class MessageCollectingInterceptor implements ChannelInterceptor {
    private final List<Message<?>> collectedMessages = new CopyOnWriteArrayList<>();

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        collectedMessages.add(message);
        return message;
    }

    public List<Message<?>> getCollectedMessages() {
        return collectedMessages;
    }
}

// 配置拦截器绑定到目标输出通道
@Configuration
public class StreamConfig {
    @Autowired
    private MessageCollectingInterceptor interceptor;

    @Bean
    public ChannelInterceptorBinding interceptorBinding() {
        return new ChannelInterceptorBinding("planeEventProducer-out-0", interceptor);
    }
}

方案3：复用类似MessageCollector的思路 #

新版Spring Cloud Stream可以通过@SpyBean监听KafkaMessageChannelBinder的发送方法，或者使用StreamBridge结合测试支持类来捕获消息，效果和旧版MessageCollector一致。

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