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);
    }
  }
}

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

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

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

Answer 1

我强烈反对这种做法。像这样的线程代码的问题是，它变得非常难以理解

应该有一个更可靠的方法

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是通量类型的字段

这实际上是每10分钟创建一个healthcheck响应值流，总是缓存最新的响应，并将其转化为热源。这意味着“在订阅之前什么都不会发生”在这种情况下不适用——流量将立即开始执行，任何线程上出现的任何新订阅都将始终获得最新结果，无论是通过还是失败handlehalthchecksuccess（）和handlehalthcheckerror（），已启动，完成当所有的都是多余的时，它们就可以去了，然后你的waitForExternalSystemUPStatus（）就可以变成一行：

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

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