Spring SCDF单处理器按数据类型分发数据至多Sink的实现问题

问题分析 #

你当前的实现方式存在核心问题：Spring Cloud Stream的函数式编程模型不支持直接返回Tuple3<Flux<String>, Flux<String>, Flux<String>>来映射到多个独立Sink。需要通过扇出路由的方式，根据数据类型将消息分发到不同的目标Sink。

正确实现方式 #

方案一：使用StreamBridge动态发送消息 #

这种方式灵活性高，可根据数据类型动态指定目标Sink的绑定名称，无需预先定义固定输出。

处理器代码实现

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.context.annotation.Bean;
import java.util.function.Function;

@Bean
public Function<Message<String>, Void> sort(StreamBridge streamBridge) {
    return message -> {
        String payload = message.getPayload();
        // 根据实际数据类型判断逻辑调整，示例通过字符串解析判断类型
        try {
            Integer.parseInt(payload);
            streamBridge.send("sink1-in-0", message);
        } catch (NumberFormatException e1) {
            try {
                Long.parseLong(payload);
                streamBridge.send("sink3-in-0", message);
            } catch (NumberFormatException e2) {
                streamBridge.send("sink2-in-0", message);
            }
        }
        return null;
    };
}

流定义命令

创建流时需指定各Sink的输入目标：

stream create --name test1 --definition "source --server.port=1001 | worker --spring.cloud.stream.function.bindings.sort-out-0=none | sink1 --spring.cloud.stream.bindings.input.destination=sink1-in-0 | sink2 --spring.cloud.stream.bindings.input.destination=sink2-in-0 | sink3 --spring.cloud.stream.bindings.input.destination=sink3-in-0"

方案二：使用多输出绑定的函数 #

通过定义函数的多个输出绑定，直接返回对应类型的消息流，逻辑更直观。

处理器代码实现

import reactor.core.publisher.Flux;
import org.springframework.context.annotation.Bean;
import java.util.function.Function;
import java.util.Map;

@Bean
public Function<Flux<String>, Map<String, Flux<String>>> sort() {
    return input -> {
        // 过滤整数类型消息
        Flux<String> integerStream = input.filter(payload -> {
            try {
                Integer.parseInt(payload);
                return true;
            } catch (NumberFormatException e) {
                return false;
            }
        });
        // 过滤长整型类型消息
        Flux<String> longStream = input.filter(payload -> {
            try {
                Long.parseLong(payload);
                return true;
            } catch (NumberFormatException e) {
                return false;
            }
        });
        // 过滤字符串类型消息
        Flux<String> stringStream = input.filter(payload -> {
            try {
                Integer.parseInt(payload);
                return false;
            } catch (NumberFormatException e1) {
                try {
                    Long.parseLong(payload);
                    return false;
                } catch (NumberFormatException e2) {
                    return true;
                }
            }
        });

        return Map.of(
            "sink1", integerStream,
            "sink2", stringStream,
            "sink3", longStream
        );
    };
}

流定义命令

将函数的每个输出绑定到对应Sink的输入：

stream create --name test1 --definition "source --server.port=1001 | worker --spring.cloud.stream.function.bindings.sort-out-sink1=sink1-in-0 --spring.cloud.stream.function.bindings.sort-out-sink2=sink2-in-0 --spring.cloud.stream.function.bindings.sort-out-sink3=sink3-in-0 | sink1 --spring.cloud.stream.bindings.input.destination=sink1-in-0 | sink2 --spring.cloud.stream.bindings.input.destination=sink2-in-0 | sink3 --spring.cloud.stream.bindings.input.destination=sink3-in-0"

关键说明 #

• 方案一的StreamBridge适合动态路由场景，无需预先定义所有输出绑定；
• 方案二的多输出绑定更直观，适合固定目标的扇出场景；
• 数据类型判断逻辑需根据实际消息格式调整（比如消息带类型头、JSON结构中的类型字段等）。

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