当请求被客户端中止时，WebFlux 如何停止发布者？答案

【问题标题】：How in WebFlux to stop publisher when request is aborted by client?当请求被客户端中止时，WebFlux 如何停止发布者？
【发布时间】：2021-06-14 16:54:47
【问题描述】：

SpringBoot v2.5.1

有一个端点请求一个长时间运行的进程结果，它是通过某种方式创建的
（为简单起见，它是Mono.fromCallable( ... long running ... )。

客户端发出请求并触发发布者完成工作，但几秒钟后客户端中止请求（即连接丢失）。并且该过程仍然继续利用资源来计算结果以丢弃。

通知 Project Reactor 的事件循环有关应该取消的不必要工作的机制是什么？

@RestController 
class EndpointSpin {
 
  @GetMapping("/spin")
  Mono<Long> spin() {
    AtomicLong counter = new AtomicLong(0);
    Instant stopTime = Instant.now().plus(Duration.of(1, ChronoUnit.HOURS));

    return Mono.fromCallable(() -> {

      while (Instant.now().isBefore(stopTime)) {
        counter.incrementAndGet();

        if (counter.get() % 10_000_000 == 0) {
          System.out.println(counter.get());
        }

        // of course this does not work
        if (Thread.currentThread().isInterrupted()){
           break;
        }
      }

      return counter.get();
    });
  }
}

【问题讨论】：

标签： java reactive-programming spring-webflux project-reactor cancellation

【解决方案1】：

fromCallable 不会阻止您在 Callable 内阻塞计算，您的示例演示了这一点。

Reactive Streams 中消除的主要方法是通过Subscription 从下游传播的cancel() 信号。

即便如此，避免在响应式代码中阻塞代码的基本要求仍然成立，因为如果运算符足够简单（即同步），阻塞步骤甚至可以阻止 cancel() 信号的传播......

在仍然收到取消通知的同时调整非反应性代码的方法是Mono.create：它公开了一个MonoSink（通过Consumer<MonoSink>），可用于将元素推送到下游，同时它有一个onCancel 处理程序。

您需要将代码重写为例如。在循环的每次迭代中检查 AtomicBoolean，并在接收器的 onCancel 处理程序中翻转 AtomicBoolean：

Mono.create(sink -> {
    AtomicBoolean isCancelled = new AtomicBoolean();
    sink.onCancel(() -> isCancelled.set(true));
    while (...) {
        ...
        if (isCancelled.get()) break;
    }
});

在您的示例中需要注意的另一件事是：AtomicInteger 是共享状态。如果您第二次订阅返回的Mono，两个订阅将共享计数器并并行递增/检查，这可能不太好。

在Mono.create 的Consumer<MonoSink> 中创建这些状态变量可确保每个订阅都有自己的单独状态。

【讨论】：