Ily*_*ich 0 java reactor project-reactor
我有一个返回的方法Mono<Output>:
interface Processor {
Mono<Output> process(Input input);
}
Run Code Online (Sandbox Code Playgroud)
我想对processor集合执行此方法:
List<Input> inputs = // get inputs
Processor processor = // get processor
List<Mono<Output>> outputs = inputs.stream().map(supplier::supply).collect(toList());
Run Code Online (Sandbox Code Playgroud)
但是List<Mono<Output>>我想得到的不是Mono<List<Output>>汇总,而是包含汇总的结果。
我尝试过reduce,但最终结果看起来很笨拙:
Mono<List<Output>> result = inputs.stream().map(processor::process)
.reduce(Mono.just(new ArrayList<>()),
(monoListOfOutput, monoOfOutput) ->
monoListOfOutput.flatMap(list -> monoOfOutput.map(output -> {
list.add(output);
return list;
})),
(left, right) ->
left.flatMap(leftList -> right.map(rightList -> {
leftList.addAll(rightList);
return leftList;
})));
Run Code Online (Sandbox Code Playgroud)
我可以用更少的代码来实现吗?
Ale*_*nov 11
// first merge all the `Mono`s:
List<Mono<Output>> outputs = ...
Flux<Output> merged = Flux.empty();
for (Mono<Output> out : outputs) {
merged = merged.mergeWith(out);
}
// then collect them
return merged.collectList();
Run Code Online (Sandbox Code Playgroud)
或(受亚历山大的回答启发)
Flux.fromIterable(outputs).flatMap(x -> x).collectList();
Run Code Online (Sandbox Code Playgroud)
如果您无需出于任何原因创建流,则可以从输入中创建Flux,对其进行映射并收集列表
Flux.fromIterable(inputs).flatMap(processor::process).collectList();
Run Code Online (Sandbox Code Playgroud)