Spring Integration中Splitter结合WebFlux OutboundGateway如何串行处理？

你的问题核心在于：WebFlux.outboundGateway是非阻塞异步的反应式组件，Splitter默认会认为下游的异步处理已经"完成"（因为Mono立即返回），所以会直接发送下一个请求，而不会等待HTTP响应返回。要实现串行处理，需要让Splitter等待每个子流程的异步操作彻底完成，或者强制下游处理串行执行。

以下是两种可行的配置方案：

方案一：使用队列通道强制串行处理 #

通过在Splitter和WebFlux网关之间添加一个QueueChannel，利用其单线程处理的特性，确保每个请求必须等到前一个请求的整个流程完成后才会被处理：

flow
    .transform(Message.class, message -> generateRequests(message))
    .split()
    .channel(c -> c.queue("serialProcessingChannel")) // 单线程队列通道，控制串行执行
    .handle( 
        WebFlux.outboundGateway(this::resolveUri, webClient)
            .httpMethod(HttpMethod.GET)
            .expectedResponseType(String.class)
    )
    .aggregate()

QueueChannel默认使用单线程的消息分发机制，每个消息会被依次处理，前一个消息的异步HTTP请求完成后，才会取出下一个消息进行处理。

方案二：将异步处理转为阻塞式（适合必须串行且可接受阻塞的场景） #

如果可以接受阻塞线程等待HTTP响应，可以直接在handle中手动调用WebClient并阻塞等待结果，这样Splitter会等到当前请求的响应返回后，再处理下一个元素：

flow
    .transform(Message.class, message -> generateRequests(message))
    .split()
    .handle((GenericHandler<Object>) (payload, headers) -> {
        // 手动构建请求并阻塞等待响应
        return webClient.get()
                .uri(resolveUri(MessageBuilder.withPayload(payload).copyHeaders(headers).build()))
                .retrieve()
                .bodyToMono(String.class)
                .block();
    })
    .aggregate()

注意：这种方法会失去WebClient的非阻塞异步特性，线程会在等待HTTP响应时被占用，适合请求量不大、必须严格串行的场景。

补充说明 #

如果你想保留反应式特性同时实现串行，可以利用Flux.concatMap操作符，确保前一个请求的Mono完成后再订阅下一个：

flow
    .transform(Message.class, message -> generateRequests(message))
    .split()
    .channel(c -> c.flux()) // 将Splitter输出转为Flux
    .transform(flux -> flux.concatMap(payload -> {
        Message<?> requestMsg = MessageBuilder.withPayload(payload).build();
        // 调用WebFlux网关并返回响应的Mono
        return WebFlux.outboundGateway(this::resolveUri, webClient)
                .httpMethod(HttpMethod.GET)
                .expectedResponseType(String.class)
                .handleMessage(requestMsg)
                .map(Message::getPayload);
    }))
    .aggregate()

concatMap会按顺序处理Flux中的每个元素，只有当前元素对应的Mono完成后，才会处理下一个元素，从而实现串行的反应式处理。

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