如何防止Mono被取消？

我正在尝试将某些内容实现为争用条件。此争用条件必须遵循以下情况：

• 同时启动两个HTTP调用。
• 返回成功完成的第一个调用的响应。
• 处理最后一个调用。(这里最重要的是，我不能丢弃最后一个呼叫，我确实需要处理它的结果：无论它的状态、成功或失败)。

此代码示例是我所实现的最接近的解决方案：

    Mono<StatusMock> monoA = webClient.get()
      .uri("https://some.url.a")
      .retrieve()
      .bodyToMono(StatusMock.class)
      .subscribeOn(Schedulers.boundedElastic());

    Mono<StatusMock> monoB = webClient.get()
      .uri("https://some.url.b")
      .retrieve()
      .bodyToMono(StatusMock.class)
      .doOnSuccess(this::verifyBody)
      .onErrorStop()
      .subscribeOn(Schedulers.boundedElastic());

    StatusMock statusMock = Flux.first(monoA, monoB)
      .blockFirst();

    if (statusMock != null) {
      return statusMock.getStatus();
    }

    return "empty";
  }

  private void verifyBody(StatusMock statusMock) {
    if (statusMock.getStatus().contains("error")) {
      log.error("throwing an exception");
      throw new RuntimeException("error");
    }
  }

public class StatusMock {

  private String status; // getters and setters implicit
}

在本例中，我使用了Flos.first方法，它对我返回第一个调用有很大帮助，但它会丢弃(取消)第二个调用，这是一个问题，因为我还需要最后一个调用的结果。

这个逻辑有什么解决方案吗？这里我使用的是Spring Project Reader，但我接受任何可以帮助我解决这种情况的库或框架。

---

解决方案

您可以在Mono上使用cache运算符来防止它们被取消：

    Mono<StatusMock> monoA = webClient.get()
      // ...
      .cache();

    Mono<StatusMock> monoB = webClient.get()
      // ...
      .cache();

    Mono.firstWithSignal(monoA, monoB);