如何防止Mono被取消?

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。第一个方法，它帮助我返回第一个调用，但它丢弃(取消)第二个调用，这是一个问题，因为我也需要最后一个调用的结果。

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

Answer 1

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

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