如何防止单声道被取消？

Question

我试图实现一些东西作为“竞赛条件”。这个竞赛条件必须遵循这些情况：

• 同时触发两个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
}

在这个例子中，我使用了Flux.first方法，它帮助我返回了第一个调用，但是它丢弃（取消）了第二个调用，这是一个问题，因为我也需要最后一个调用的结果。

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

Answer 1

您可以使用Mono上的cache操作符来防止它们被取消：

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

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

    Mono.firstWithSignal(monoA, monoB);