Reactive Java中如何将Flux批量调用结果收集为Key-Result Map

你当前的写法使用外部可变集合收集结果不符合响应式编程的最佳实践，我们可以直接通过Reactor内置的聚合操作符实现键值对结果的收集，调整代码如下：

import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

// 错误结果集合保留原有逻辑即可
Map<String, Throwable> erroredResults = new ConcurrentHashMap<>();
// 直接通过响应式操作符聚合得到成功结果的HashMap
Map<String, GetResult> successfulResults = Flux.fromIterable(docsToFetch)
    .flatMap(key -> reactiveCollection.getAndTouch(key, Duration.ofMinutes(extendExpiryInMin))
        // 成功查询后将key和结果配对为二元组，保留关联关系
        .map(result -> Tuples.of(key, result))
        .onErrorResume(e -> {
            erroredResults.put(key, e);
            return Mono.empty();
        })
    )
    // 原生操作符直接聚合为HashMap，天然线程安全无需手动维护同步集合
    .collectMap(Tuple2::getT1, Tuple2::getT2)
    .block();

可选兼容方案 #

如果你不想引入Reactor的Tuple工具类，也可以用JDK自带的MapEntry实现，对应调整两处代码即可：

// 配对键值对改用SimpleEntry
.map(result -> new AbstractMap.SimpleEntry<>(key, result))
// 收集时取Entry的键值
.collectMap(Map.Entry::getKey, Map.Entry::getValue)

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