Spring Cloud Stream多通道条件路由：多主题独立处理问题

问题背景 #

使用Spring Cloud Stream Kafka函数式编程模式，存在两个主题：

• Topic1：包含Car、Bike类型消息，需分别通过processCar、processBike函数处理
• Topic2：包含Fruits、Vegetables类型消息，需分别通过processFruits、processVegetables函数处理

已通过MessageRoutingCallback实现Topic1的消息路由，但无法对Topic2实现相同功能，且由于两个Topic使用不同绑定器、故障处理需求不同，不能将它们配置到同一个functionRouter-in-0的destination中。

当前代码：

@Bean
public MessageRoutingCallback customRouter() {
    String EVENT_TYPE = "eventType";

    Map<String, String> evenTypeToFunctionNameMap = new HashMap<>();
    evenTypeToFunctionNameMap.put(
            "car",
            "processCar"
    );
    evenTypeToFunctionNameMap.put(
            "bike",
            "processBike"
    );
    

    return new MessageRoutingCallback() {
        @Override
        public FunctionRoutingResult routingResult(Message<?> message) {
            return new FunctionRoutingResult(
                    evenTypeToFunctionNameMap.get(
                            message.getHeaders().get(EVENT_TYPE, String.class)
                    )
            );
        }
    };
}

当前配置：

spring.cloud.stream.bindings.functionRouter-in-0.destination=topic1
spring.cloud.stream.bindings.functionRouter-in-0.binder=default

解决方案 #

核心思路是为每个Topic创建独立的functionRouter输入绑定，然后在路由回调中根据消息来源的输入通道区分处理逻辑。

1. 添加第二个functionRouter输入绑定配置 #

新增Topic2的绑定配置，指定对应绑定器（示例中为binder2）：

# Topic1原有配置
spring.cloud.stream.bindings.functionRouter-in-0.destination=topic1
spring.cloud.stream.bindings.functionRouter-in-0.binder=default

# Topic2新增配置
spring.cloud.stream.bindings.functionRouter-in-1.destination=topic2
spring.cloud.stream.bindings.functionRouter-in-1.binder=binder2

2. 修改MessageRoutingCallback逻辑 #

通过MessageHeaders.RECEIVED_CHANNEL header识别消息来源的输入通道，结合eventType路由到对应处理函数，还可添加通道与事件类型的匹配校验：

@Bean
public MessageRoutingCallback customRouter() {
    String EVENT_TYPE = "eventType";
    String RECEIVED_CHANNEL = MessageHeaders.RECEIVED_CHANNEL;

    // 构建完整的事件类型到处理函数的映射
    Map<String, String> eventTypeToFunctionNameMap = new HashMap<>();
    // Topic1相关映射
    eventTypeToFunctionNameMap.put("car", "processCar");
    eventTypeToFunctionNameMap.put("bike", "processBike");
    // Topic2相关映射
    eventTypeToFunctionNameMap.put("fruits", "processFruits");
    eventTypeToFunctionNameMap.put("vegetables", "processVegetables");

    return new MessageRoutingCallback() {
        @Override
        public FunctionRoutingResult routingResult(Message<?> message) {
            String eventType = message.getHeaders().get(EVENT_TYPE, String.class);
            String receivedChannel = message.getHeaders().get(RECEIVED_CHANNEL, String.class);

            // 按通道做事件类型校验，避免跨Topic消息错误路由
            if ("functionRouter-in-1".equals(receivedChannel)) {
                if (!Arrays.asList("fruits", "vegetables").contains(eventType)) {
                    // 返回null可触发预设的错误处理逻辑
                    return new FunctionRoutingResult(null);
                }
            } else {
                if (!Arrays.asList("car", "bike").contains(eventType)) {
                    return new FunctionRoutingResult(null);
                }
            }

            return new FunctionRoutingResult(eventTypeToFunctionNameMap.get(eventType));
        }
    };
}

3. 定义处理函数Bean #

确保四个处理函数以Spring Bean形式存在，示例如下：

@Bean
public Consumer<Car> processCar() {
    return car -> {
        // 实现Car类型消息处理逻辑
    };
}

@Bean
public Consumer<Bike> processBike() {
    return bike -> {
        // 实现Bike类型消息处理逻辑
    };
}

@Bean
public Consumer<Fruits> processFruits() {
    return fruits -> {
        // 实现Fruits类型消息处理逻辑
    };
}

@Bean
public Consumer<Vegetables> processVegetables() {
    return vegetables -> {
        // 实现Vegetables类型消息处理逻辑
    };
}

4. 独立配置故障处理（可选） #

针对两个输入通道配置不同的故障处理策略：

# Topic1故障处理配置
spring.cloud.stream.bindings.functionRouter-in-0.consumer.max-attempts=3
spring.cloud.stream.bindings.functionRouter-in-0.consumer.back-off-initial-interval=1000

# Topic2故障处理配置
spring.cloud.stream.bindings.functionRouter-in-1.consumer.max-attempts=5
spring.cloud.stream.bindings.functionRouter-in-1.consumer.back-off-initial-interval=2000

原理说明 #

Spring Cloud Stream的functionRouter支持多输入绑定（in-0、in-1...），每个绑定可关联不同Topic与绑定器。通过RECEIVED_CHANNEL header可精准识别消息来源通道，结合业务eventType即可实现跨Topic的独立路由，同时满足不同绑定器与故障处理的差异化需求。

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