Spring Boot中Kafka消息路由问题：sendData转至sendDataA/B通道

你的方案失效的核心原因有几个：

• YAML配置的通道名称与代码不匹配：代码中定义的是sendDataA/sendDataB，但配置里写的是sendDataOne/sendDataTwo，导致Spring Cloud Stream无法将通道正确绑定到对应Kafka主题。
• sendData通道被Spring Cloud Stream直接绑定到了test-topic，自定义的IntegrationFlow并未拦截到该通道的消息——Stream的绑定逻辑会自动为通道创建出站适配器，消息直接被发送到Kafka，不会进入你的路由流程。
• 手动定义的sendDataA/sendDataB MessageChannel并未被Spring Cloud Stream绑定到Kafka主题，只有Stream自身管理的通道才能完成主题映射。

1. 修正YAML配置 #

对齐通道名称，同时将sendData设为内部通道（不配置destination，避免Stream直接绑定到Kafka）：

spring:
  cloud:
    stream:
      bindings:
        sendDataA:
          destination: test-topic-1
          binder: kafka
        sendDataB:
          destination: test-topic-2
          binder: kafka

2. 调整IntegrationFlow配置 #

移除手动定义的sendDataA/sendDataB通道（由Spring Cloud Stream自动创建），确保sendData作为内部通道，消息进入路由流程后转发到Stream管理的出站通道：

@Configuration
@EnableIntegration
public class OutboundChannels {

    @Autowired
    private RoundRobinRouter roundRobinRouter;

    @Bean(name = "sendData")
    public MessageChannel sendData() {
        return MessageChannels.direct().get();
    }

    @Bean
    public IntegrationFlow routeRoundRobinFlow() {
        return IntegrationFlows.from("sendData")
                .route(roundRobinRouter, "route",
                        r -> r.channelMapping("sendDataA", "sendDataA")
                              .channelMapping("sendDataB", "sendDataB"))
                .get();
    }
}

3. 实现真正的轮询路由逻辑 #

修改RoundRobinRouter，用原子计数器实现轮询逻辑：

@Component
class RoundRobinRouter {
    private final AtomicInteger counter = new AtomicInteger(0);

    public String route(Object payload) {
        return counter.getAndIncrement() % 2 == 0 ? "sendDataA" : "sendDataB";
    }
}

4. 确认依赖配置 #

确保项目引入spring-cloud-starter-stream-kafka依赖，且Spring Cloud Stream版本与Kafka版本兼容。

如果使用Spring Cloud Stream 3.x及以上版本，可采用更简洁的函数式模型实现：

1. 定义路由处理函数：

@Component
public class RouteFunction {
    private final AtomicInteger counter = new AtomicInteger(0);

    @Bean
    public Function<Message<String>, Message<String>> routeToTopics() {
        return message -> {
            String targetChannel = counter.getAndIncrement() % 2 == 0 ? "sendDataA" : "sendDataB";
            return MessageBuilder.fromMessage(message)
                    .setHeader(MessageHeaders.CHANNEL, targetChannel)
                    .build();
        };
    }
}

2. 对应YAML配置：

spring:
  cloud:
    stream:
      function:
        definition: routeToTopics
      bindings:
        routeToTopics-in-0:
          destination: sendData
        sendDataA:
          destination: test-topic-1
          binder: kafka
        sendDataB:
          destination: test-topic-2
          binder: kafka

调用时只需将消息发送到routeToTopics-in-0通道即可完成路由。

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