Flux中reduce是否等待flatMap所有元素完成？如何确保全量处理后执行操作？

核心疑问：reduce是否会等待所有元素处理完成？ #

会。reduce是Reactor的终端操作，它的特性就是要收集上游Flux产生的全部元素后，才会执行合并逻辑生成最终的Mono<Result>。也就是说，只有当flatMap里所有getUserItems(userId)的异步操作都执行完毕、所有doSomethingWithUserItemRecord的结果都输出后，reduce才会开始工作，之后才会进入map执行批量用户记录操作。

额外说明 #

只要getUserItems(userId)返回的流最终会正常结束（不会无限阻塞），当前代码的逻辑就已经满足你“所有用户条目操作完成后再处理用户记录”的需求。

如果想要更直观地确认所有结果都已收集，也可以把reduce替换成collectList()先收集所有doSomethingWithUserItemRecord的结果，再手动合并：

Flux.fromIterable(userIds)
    .flatMap(userId -> getUserItems(userId)
                        .map(userItemIds -> doSomethingWithUserItemRecord(userId, userItemIds)))
    .collectList()
    .map(itemResults -> {
        Result result = new Result(0, true);
        for (Result itemResult : itemResults) {
            result = mergeResult(result, itemResult);
        }
        return result;
    })
    .map(result -> doSomethingWithUsersRecords(result))

这种写法和原代码的reduce逻辑完全等价，只是更显式地展示了“先收集所有结果再合并”的过程。

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