如何按顺序执行 Mono 列表

Question

我有一个我想处理的单声道列表，但它们必须按顺序执行，并且只有在前一个单声道完成后才应该执行下一个。

private Mono<List<Result>> processGoals(List<> goals,Data data) {
    
    List<Mono<Result>> plans = goals
                              .stream()
                              .map(plan -> processGoal(plan, data))
                              .collect(Collectors.toList());

}

我尝试使用

return Flux.concat(plans).subscribeOn(Schedulers.single()).collectList();

但这会在前一个单声道完成之前执行下一个单声道。

Answer 1

.flatMapSequential(goal -> processGoal(goal, data), 1)

最后一个参数并发1很重要。我之前尝试过没有并发参数，但没有奏效。感谢@Michael McFadyen

Answer 2

Flux#concatMap 是这种情况下的最佳选择。

它将按顺序合并每个映射的发布者并一次触发一个，而无需显式定义concurrency 参数。

这是一个完整的例子：

Flux.fromIterable(goals))
        .concatMap(goal -> processGoal(goal, data))
        .collectList();

Answer 3

而不是混合两种范式：Streams 和 Reactive Streams。您可以尝试完全反应。

尝试以下操作：

Mono<List<Result>> res = Flux.just(goals.toArray(Goal[]::new))
                            .flatMapSequential(goal -> processGoal(goal, data))
                            .subscribeOn(Schedulers.single())
                            .collectList();