替换默认 Schedulers

Doubao 中英对照 Replacing Default Schedulers Replacing Default Schedulers

正如我们在线程与调度器一节中所描述的，Reactor Core 自带多种 Scheduler 实现。虽然您随时可以通过 new* 工厂方法创建新实例，但每种 Scheduler 风格还有一个默认单例实例，可通过直接工厂方法获取（例如 Schedulers.boundedElastic() 与 Schedulers.newBoundedElastic(…) 的区别）。

当您没有显式指定 Scheduler 时，那些需要 Scheduler 才能工作的操作符所使用的，正是这些默认实例。例如，Flux#delayElements(Duration) 使用的就是 Schedulers.parallel() 实例。

但在某些情况下，您可能需要以一种横切的方式把这些默认实例替换成别的实现，而不必逐一确认所调用的每个操作符都把您指定的 Scheduler 作为参数传入。一个例子是：为了做指标埋点，通过包装真实的调度器来测量每个被调度任务的耗时。换句话说，您可能想要替换默认的 Schedulers。

可以通过 Schedulers.Factory 类来替换默认调度器。默认情况下，Factory 通过同名方法创建所有标准 Scheduler。您可以用自定义实现覆盖其中每一个方法。

此外，该工厂还暴露了一个额外的定制方法：decorateExecutorService。每创建一个由 ScheduledExecutorService 支撑的 Reactor Core Scheduler（即使是非默认实例，例如通过调用 Schedulers.newParallel() 创建的实例）时，都会调用该方法。

这让您可以调整将要使用的 ScheduledExecutorService：默认的执行器服务以 Supplier 的形式暴露出来，根据正在配置的 Scheduler 的类型，您可以选择完全绕过该 supplier 并返回自己的实例，也可以 get() 出默认实例再对其进行包装。

important

一旦创建了满足您需求的 Factory，就必须通过调用 Schedulers.setFactory(Factory) 来安装它。

最后，Schedulers 中还有一个可定制的钩子：onHandleError。每当提交到 Scheduler 的 Runnable 任务抛出 Exception 时，都会调用这个钩子（注意：如果运行该任务的 Thread 设置了 UncaughtExceptionHandler，那么该处理器和此钩子都会被调用）。