线程与调度器

Doubao 中英对照 Threading and Schedulers

目录

• 1. publishOn 方法
• 2. subscribeOn 方法

与 RxJava 一样，Reactor 可以被视为并发无关的（concurrency-agnostic）。也就是说，它不强制采用某种并发模型，而是把主动权交给您——开发者。然而，这并不妨碍该库在并发方面为您提供帮助。

获取一个 Flux 或 Mono 并不一定意味着它会在一个专用的 Thread 中运行。相反，大多数操作符会在前一个操作符执行所在的 Thread 中继续工作。除非另行指定，最顶层的操作符（即源）本身运行在发起 subscribe() 调用的那个 Thread 中。以下示例在一个新线程中运行一个 Mono：

public static void main(String[] args) throws InterruptedException {
  final Mono<String> mono = Mono.just("hello "); 1

  Thread t = new Thread(() -> mono
      .map(msg -> msg + "thread ")
      .subscribe(v -> 2
          System.out.println(v + Thread.currentThread().getName()) 3
      )
  );
  t.start();
  t.join();

}

• 该 Mono<String> 在线程 main 中组装。

• 然而，它是在线程 Thread-0 中被订阅的。

• 因此，map 和 onNext 回调实际上都运行在 Thread-0 中

前述代码产生如下输出：

hello thread Thread-0

在 Reactor 中，执行模型以及执行发生的位置由所使用的 Scheduler 决定。一个 Scheduler 具有类似于 ExecutorService 的调度职责，但拥有专门的抽象让它能做更多事情，特别是充当时钟，并支持更广泛的实现（用于测试的虚拟时间、蹦床式调度或立即调度等等）。

Schedulers 类提供了若干静态方法，让您可以访问以下执行上下文：

• 没有执行上下文（Schedulers.immediate()）：在处理时，提交的 Runnable 会被直接执行，实际上就是在当前 Thread 上运行它们（可视为一个“空对象”或空操作的 Scheduler）。

• 单个可复用线程（Schedulers.single()）。请注意，该方法会为所有调用者复用同一个线程，直到该调度器被处置。如果您想要每次调用都有专用线程，请在每次调用时使用 Schedulers.newSingle()。

• 一个无界的弹性线程池（Schedulers.elastic()）。随着 Schedulers.boundedElastic() 的引入，已不再推荐使用它，因为它倾向于掩盖背压问题并导致线程过多（见下文）。

• 一个有界的弹性线程池（Schedulers.boundedElastic()）。这是一种便捷方式，可为阻塞进程分配它自己的线程，使其不会占用其他资源。对于 I/O 阻塞型工作，这是更好的选择。请参阅如何包装同步阻塞调用？，但它不会用新建线程给系统造成太大压力。从 3.6.0 起，根据配置不同，它可以提供两种不同的实现：

  • 基于 ExecutorService，在任务之间复用平台线程。与它的前身 elastic() 一样，该实现会根据需要创建新的工作线程池并复用空闲的线程池。空闲过久（默认 60 秒）的工作线程池也会被处置。与前身 elastic() 不同，它对可创建的后台线程数量设有上限（默认为 CPU 核数 × 10）。达到上限后提交的多达 100 000 个任务会进入队列，并在有线程可用时重新被调度（如果是带延迟调度，则延迟从线程可用时开始计算）。

  • 基于每任务一线程，设计为运行在 VirtualThread 实例上。要使用该功能，应用应运行在 Java 21+ 环境中，并将系统属性 reactor.schedulers.defaultBoundedElasticOnVirtualThreads 设为 true。完成上述设置后，共享的 Schedulers.boundedElastic() 会返回一个特定的 BoundedElasticScheduler 实现，专为在新的 VirtualThread 类实例上运行每个任务而定制。该实现在行为上与基于 ExecutorService 的实现相似，但没有空闲线程池，并为每个任务创建一个新的 VirtualThread。

• 一个为并行工作调优的固定工作线程池（Schedulers.parallel()）。它创建的工作线程数量与您的 CPU 核数相同。

此外，您可以通过 Schedulers.fromExecutorService(ExecutorService) 从任意已有的 ExecutorService 创建一个 Scheduler。（您也可以从一个 Executor 创建一个，但不推荐这样做。）

您还可以使用 newXXX 方法创建各种调度器类型的新实例。例如，Schedulers.newParallel(yourScheduleName) 会创建一个名为 yourScheduleName 的新并行调度器。

注意

虽然 boundedElastic 是为在无法避免时协助遗留阻塞代码而设计的，但 single 和 parallel 并非如此。因此，在默认的 single 和 parallel 调度器内部使用 Reactor 阻塞 API（block()、blockFirst()、blockLast()，以及遍历 toIterable() 或 toStream()）会抛出 IllegalStateException。

自定义 Schedulers 也可以通过创建实现 NonBlocking 标记接口的 Thread 实例，将其标记为“仅非阻塞”。

有些操作符默认使用 Schedulers 中的某个特定调度器（通常也会让您选择提供另一个调度器）。例如，调用 Flux.interval(Duration.ofMillis(300)) 工厂方法会产生一个每 300ms 发出一次节拍的 Flux<Long>。默认情况下，这由 Schedulers.parallel() 启用。下面这行代码将调度器改为一个类似于 Schedulers.single() 的新实例：

Flux.interval(Duration.ofMillis(300), Schedulers.newSingle("test"))

Reactor 提供了两种在响应式链中切换执行上下文（或 Scheduler）的方式：publishOn 和 subscribeOn。二者都接受一个 Scheduler，并让您把执行上下文切换到该调度器。但 publishOn 在链中的位置很关键，而 subscribeOn 的位置则无关紧要。要理解这种差异，您首先要记住在订阅之前什么都不会发生。

在 Reactor 中，当您链接操作符时，可以根据需要把任意多个 Flux 和 Mono 实现层层嵌套。一旦订阅，就会创建一条由 Subscriber 对象组成的链，沿链向后（向上）延伸到第一个发布者。这一过程对您实际上是透明隐藏的。您所能看到的只是最外层的 Flux（或 Mono）和 Subscription，但真正的工作发生在这些中间的、操作符专属的订阅者之中。

了解这一点之后，我们可以更仔细地看看 publishOn 和 subscribeOn 这两个操作符：

1. publishOn 方法

publishOn 的应用方式与其他任何操作符一样，位于订阅者链的中间。它从上游获取信号，并在关联 Scheduler 的某个 worker 上执行回调，同时把信号重放到下游。因此，它会影响其后各操作符的执行位置（直到链中再接入另一个 publishOn），如下所示：

• 把执行上下文切换为由 Scheduler 挑选的某个 Thread

• 按照规范，onNext 调用是顺序发生的，因此这只会占用单个线程

• 除非它们作用于某个特定 Scheduler，否则 publishOn 之后的操作符会在同一个线程上继续执行

以下示例使用 publishOn 方法：

Scheduler s = Schedulers.newParallel("parallel-scheduler", 4); 1

final Flux<String> flux = Flux
    .range(1, 2)
    .map(i -> 10 + i)  2
    .publishOn(s)  3
    .map(i -> "value " + i);  4

new Thread(() -> flux.subscribe(System.out::println));  5

• 创建一个由四个 Thread 实例支撑的新 Scheduler。

• 第一个 map 运行在 <5> 中的匿名线程上。

• publishOn 把整条序列切换到从 <1> 挑选的某个 Thread 上。

• 第二个 map 运行在来自 <1> 的 Thread 上。

• 这个匿名 Thread 是发生订阅的地方。打印发生在最新的执行上下文中，也就是 publishOn 所在的那个上下文。

2. subscribeOn 方法

subscribeOn 作用于订阅过程，即向后链被构建之时。通常建议把它紧接在数据源之后放置，因为中间操作符可能影响执行的上下文。

然而，这并不影响后续 publishOn 调用的行为——它们仍然会为其后的那部分链切换执行上下文。

• 改变操作符整条链发起订阅所用的 Thread

• 从 Scheduler 中挑选一个线程

备注

下游链中距离最近的 subscribeOn 调用，才真正负责把订阅信号和请求信号调度到源或能拦截它们的操作符（doFirst、doOnRequest）。使用多个 subscribeOn 调用只会带来没有价值的、不必要的线程切换。

以下示例使用 subscribeOn 方法：

Scheduler s = Schedulers.newParallel("parallel-scheduler", 4); 1

final Flux<String> flux = Flux
    .range(1, 2)
    .map(i -> 10 + i)  2
    .subscribeOn(s)  3
    .map(i -> "value " + i);  4

new Thread(() -> flux.subscribe(System.out::println));  5

• 创建一个由四个 Thread 支撑的新 Scheduler。

• 第一个 map 运行在这四个线程之一上……

• ……因为 subscribeOn 从订阅时（<5>）起就切换了整条序列。

• 第二个 map 也运行在同一个线程上。

• 这个匿名 Thread 是最初发生订阅的地方，但 subscribeOn 立即把它转移到四个调度器线程之一。