基于Java中可变属性返回Mono的方法的线程安全

Question

在我的 Spring Boot 应用程序中，我有一个组件应该监控另一个外部系统的健康状态。该组件还提供了一个公共方法，反应链可以订阅以等待外部系统启动。

@Component
public class ExternalHealthChecker {
  private static final Logger LOG = LoggerFactory.getLogger(ExternalHealthChecker.class);

  private final WebClient externalSystemWebClient = WebClient.builder().build(); // config omitted

  private volatile boolean isUp = true;
  private volatile CompletableFuture<String> completeWhenUp = new CompletableFuture<>();

  @Scheduled(cron = "0/10 * * ? * *")
  private void checkExternalSystemHealth() {
    webClient.get() //
        .uri("/health") //
        .retrieve() //
        .bodyToMono(Void.class) //
        .doOnError(this::handleHealthCheckError) //
        .doOnSuccess(nothing -> this.handleHealthCheckSuccess()) //
        .subscribe(); //
  }

  private void handleHealthCheckError(final Throwable error) {
    if (this.isUp) {
      LOG.error("External System is now DOWN. Health check failed: {}.", error.getMessage());
    }
    this.isUp = false;
  }

  private void handleHealthCheckSuccess() {
  // the status changed from down -> up, which has to complete the future that might be currently waited on  
  if (!this.isUp) {
      LOG.warn("External System is now UP again.");
      this.isUp = true;
      this.completeWhenUp.complete("UP");
      this.completeWhenUp = new CompletableFuture<>();
    }
  }


  public Mono<String> waitForExternalSystemUPStatus() {
    if (this.isUp) {
      LOG.info("External System is already UP!");
      return Mono.empty();
    } else {
      LOG.warn("External System is DOWN. Requesting process can now wait for UP status!");
      return Mono.fromFuture(completeWhenUp);
    }
  }
}

waitForExternalSystemUPStatus 方法是公共的，可以从许多不同的线程调用。这背后的想法是为应用程序中的一些反应通量链提供一种暂停其处理直到外部系统启动的方法。当外部系统宕机时，这些链无法处理它们的元素。

someFlux
  .doOnNext(record -> LOG.info("Next element")
  .delayUntil(record -> externalHealthChecker.waitForExternalSystemUPStatus())
  ... // starting processing

这里的问题是我无法真正理解这段代码的哪一部分需要同步。我认为多个线程同时调用waitForExternalSystemUPStatus不应该有问题，因为这种方法没有写任何东西。所以我觉得这个方法不需要同步。但是，带有@Scheduled 注释的方法也将在它自己的线程上运行，并且实际上会写入isUp 的值，并且还可能将completeWhenUp 的引用更改为新的未完成的未来实例。我已经用volatile标记了这两个可变属性，因为在阅读Java 中的这个关键字后，我觉得它有助于确保读取这两个值的线程看到最新的值。但是，我不确定是否还需要在部分代码中添加 synchronized 关键字。我也不确定 synchronized 关键字是否与反应器代码配合得很好，我很难找到这方面的信息。也许还有一种方法可以以更完整、更被动的方式提供ExternalHealthChecker 的功能，但我想不出任何方法。

Answer 1

我强烈建议不要使用这种方法。像这样的线程代码的问题是它变得非常难以理解和推理。我认为您至少需要同步 handleHealthCheckSuccess() 和 waitForExternalSystemUPStatus() 中引用您的 completeWhenUp 字段的部分，否则您可能会遇到种族危险（只有一个人写信给它，但它可能会在写入之后被乱序读取）——但很可能还有其他我遗漏的东西，如果是这样，它可能会显示为这些恼人的“百万分之一”类型的错误之一，这几乎是不可能的固定。

应该有一个更可靠和简单的方法来实现这一点。当您的ExternalHealthChecker 组件按如下方式创建时，我将创建一个通量，而不是使用 Spring 调度程序：

healthCheckStream = Flux.interval(Duration.ofMinutes(10))
        .flatMap(i ->
                webClient.get().uri("/health")
                        .retrieve()
                        .bodyToMono(String.class)
                        .map(s -> true)
                        .onErrorResume(e -> Mono.just(false)))
        .cache(1);

...其中healthCheckStream 是Flux<Boolean> 类型的字段。 （注意它不需要是易变的，因为你永远不会替换它，所以跨线程的担忧不适用 - 它是相同的流，将根据运行状况检查状态每 10 分钟更新一次不同的结果，无论线程您将从中访问它。）

这实质上每 10 分钟创建一个健康检查响应值流，始终缓存最新响应，并将其变成热源。这意味着“在您订阅之前什么都不会发生”在这种情况下不适用 - 通量将立即开始执行，并且任何进入任何线程的新订阅者将始终获得最新结果，无论是通过还是失败. handleHealthCheckSuccess() 和 handleHealthCheckError()、isUp 和 completeWhenUp 都是多余的，它们可以删除 - 然后你的 waitForExternalSystemUPStatus() 可以变成一行：

return healthCheckStream.filter(x -> x).next();

...然后工作完成，您可以从任何地方调用它，您将拥有一个 Mono，它只会在系统启动时完成。