等待运行 Reactor Mono 实例完成

Question

我写这段代码是为了分拆大量的WebClients（受reactor.ipc.netty.workerCount限制），立即启动Mono，等待所有的Monos完成：

   List<Mono<List<MetricDataModel>>> monos = new ArrayList<>(metricConfigs.size());
   for (MetricConfig metricConfig : metricConfigs) {
        try {
            monos.add(extractMetrics.queryMetricData(metricConfig)
                  .doOnSuccess(result -> {
                      metricDataList.addAll(result);
                  })
                  .cache());
        } catch (Exception e) {
        }
    }

    Mono.when(monos)
          .doFinally(onFinally -> {
              Map<String, Date> latestMap;
              try {
                  latestMap = extractInsights.queryInsights();
                  Transform transform = new Transform(copierConfig.getEventType());
                  ArrayList<Event> eventList = transform.toEvents(latestMap, metricDataList);
              } catch (Exception e) {
                  log.error("copy: mono: when: {}", e.getMessage(), e);
              }
          })
          .block();

它“有效”，即结果符合预期。

两个问题：

1. 这是正确吗？ cache() 是否会导致 when 等待所有 Monos 完成？
2. 它高效吗？有没有办法让这更快？

谢谢。

Answer 1

您应该尽可能地尝试：

• 使用 Reactor 运算符并组成单个反应链
• 避免将doOn* 运算符用于除副作用（如日志记录）以外的其他事情
• 避免共享状态

您的代码可能看起来更像

List<MetricConfig> metricConfigs = //...
Mono<List<MetricDataModel>> data = Flux.fromIterable(metricConfigs)
    .flatMap(config -> extractMetrics.queryMetricData(config))
    .collectList();

此外，cache() 运算符不会等待流完成（这实际上是 then() 的工作）。