问题根源 #

你遇到的核心问题是内部路由通道被意外绑定到了Kafka默认生产者，导致消息被轮询分发到两个订阅者：

1. 你定义的HeaderValueRouter（负责路由到动态目标通道）
2. Spring Cloud Stream自动创建的默认生产者（直接发送到默认Kafka主题）

由于OUTPUT_CHANNEL默认是DirectChannel（点对点通道），消息会被轮询分配给这两个订阅者，所以10条消息中一半被正确路由，另一半直接进入默认通道。

另外，你在router()方法上同时使用@ServiceActivator和@Bean属于不规范用法，虽然不是直接诱因，但可能引发其他潜在问题。

修复方案 #

1. 调整配置类，移除内部通道的@Output绑定

将DynamicSource接口移除（或修改为空），因为内部路由通道不需要绑定到外部Kafka，动态通道由BinderAwareChannelResolver自动创建：

@EnableBinding // 无需指定绑定接口，动态通道由Resolver创建
public class CloudStreamConfig {
    @Autowired
    private BinderAwareChannelResolver resolver;
    public static final String CHANNEL_HEADER = "channelHeader";
    public static final String OUTPUT_CHANNEL = "outputChannel";
    private final String defaultChannel = "defaultChannel";

    // 显式定义内部路由通道（可选，Spring会自动创建同名通道）
    @Bean
    public MessageChannel outputChannel() {
        return new DirectChannel();
    }

    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public HeaderValueRouter router() {
        HeaderValueRouter router = new HeaderValueRouter(CHANNEL_HEADER);
        router.setDefaultOutputChannelName(defaultChannel);
        router.setChannelResolver(resolver);
        return router;
    }
}

2. 修正控制器的通道注入

确保注入的是内部路由通道，而非之前的绑定通道：

@Autowired
@Qualifier(CloudStreamConfig.OUTPUT_CHANNEL)
private MessageChannel localChannel;

3. （可选）规范路由处理器的定义

将@Bean和@ServiceActivator分离，让代码结构更清晰：

@Bean
public HeaderValueRouter dynamicRouter() {
    HeaderValueRouter router = new HeaderValueRouter(CHANNEL_HEADER);
    router.setDefaultOutputChannelName(defaultChannel);
    router.setChannelResolver(resolver);
    return router;
}

@ServiceActivator(inputChannel = OUTPUT_CHANNEL, handler = "dynamicRouter")
public void routeMessages() {
    // 无需实现逻辑，handler指定为上面定义的路由bean
}

关键概念说明 #

• BinderAwareChannelResolver：它的核心作用是动态创建与外部绑定器（如Kafka）绑定的通道。当调用resolveDestination(channelName)时，会自动生成对应名称的Kafka主题绑定通道，无需提前在@EnableBinding中定义。
• @Output的正确用法：仅用于标记需要直接绑定到外部消息中间件的通道，内部路由通道不需要添加该注解，否则Spring Cloud Stream会自动为其创建默认生产者，导致消息被分流。

修改完成后，所有发送到localChannel的消息都会通过HeaderValueRouter统一路由，不会再出现消息分流到默认通道的问题。

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