java

6.3 自定义调度器与 Worker 模型

从 Scheduler/Worker 的真实接口契约出发，手写一个可运行的 LoggingScheduler：直接调度与 Worker 串行调度的区别、Worker.dispose 的取消语义、Scheduler.dispose 对 Worker 的回收，以及延迟/周期调度默认被拒绝的原因。再用 reactor-test 的 VirtualTimeScheduler 演示 StepVerifier 虚拟时间与手动 advanceTimeBy 两种测试方式，并给出按 key 分区和包装现有调度器的完整实现。示例基于 reactor-core 3.8.8、reactor-test 3.8.8。

Leeting Yan 9 分钟阅读 4250 字

先别急着写：大多数场景已经有现成调度器

自定义 Scheduler 的维护成本很高：要自己保证串行性、取消语义、生命周期和拒绝策略。下面这些需求用工厂方法就能覆盖：

需求现成入口关键行为
独立单线程、严格串行Schedulers.newSingle("name")默认非守护线程，必须 dispose() 才能退出 VM
固定数量的计算线程Schedulers.newParallel("name", n)线程标记为非阻塞，对不会立即完成的源调用 block() 会抛 IllegalStateException
有界阻塞任务池Schedulers.newBoundedElastic(...)第 6.2 节的上限与队列
包装现成 ExecutorServiceSchedulers.fromExecutorService(es, "name")dispose() 会 shutdownNow() 这个线程池
包装普通 ExecutorSchedulers.fromExecutor(executor)不保证 Worker 内 FIFO/串行；取消后任务仍留在队列里
测试时间相关操作VirtualTimeScheduler（reactor-test）手动或由 StepVerifier 推进虚拟时钟

只有需要「按 key 路由到固定线程」「给每次任务统一打点」这类额外行为时，才值得写装饰器或自定义实现。手写之前，先把接口契约看清楚。

Scheduler 与 Worker 的真实接口

很多资料把接口简化成两个方法，实际签名更完整（省略方法体细节）：

public interface Scheduler extends Disposable {

    Disposable schedule(Runnable task);   // 唯一必须实现的调度方法

    default Disposable schedule(Runnable task, long delay, TimeUnit unit) {
        throw Exceptions.failWithRejectedNotTimeCapable();
    }

    default Disposable schedulePeriodically(Runnable task,
            long initialDelay, long period, TimeUnit unit) {
        throw Exceptions.failWithRejectedNotTimeCapable();
    }

    default long now(TimeUnit unit) { /* 默认用 System.currentTimeMillis/nanoTime */ }

    Worker createWorker();

    @Override default void dispose() { }  // 默认空实现
    default Mono<Void> disposeGracefully() { /* 默认调用 dispose() */ }
    @Deprecated default void start() { }
    default void init() { start(); }

    interface Worker extends Disposable {
        Disposable schedule(Runnable task);

        default Disposable schedule(Runnable task, long delay, TimeUnit unit) {
            throw Exceptions.failWithRejectedNotTimeCapable();
        }

        default Disposable schedulePeriodically(Runnable task,
                long initialDelay, long period, TimeUnit unit) {
            throw Exceptions.failWithRejectedNotTimeCapable();
        }
    }
}

有几个容易被忽略的点：

  • Scheduler 和 Worker 都继承 Disposable，所以 isDisposed() 是实现类的必选方法。只写 schedule 和 createWorker 编译不过。
  • 延迟和周期调度是默认方法，默认实现直接抛 RejectedExecutionException。实测异常类型是内部子类 Exceptions$StaticRejectedExecutionException，消息为 Scheduler is not capable of time-based scheduling。
  • 如果拿这种「只会立即调度」的调度器去跑 Flux.interval(..., scheduler)，错误会被包装成 Exceptions$ReactorRejectedExecutionException: Scheduler unavailable。
  • Scheduler.dispose() 有默认空实现，不代表可以不管生命周期；自定义实现必须自己关闭线程资源。
  • start() 已标 @Deprecated，新实现应关注 init()。

两者在操作符链里的分工也不同：

Scheduler.scheduleWorker.schedule
执行位置调度器的任意可用执行槽该 Worker 专属的串行上下文
顺序接口明确写着 no ordering guaranteesReactor 实现通常 FIFO；接口措辞是 “should usually”
取消返回的 Disposable 取消单个任务单个任务的 Disposable，外加 worker.dispose() 取消该 Worker 的全部任务
谁在用subscribeOnpublishOn 等需要串行边界的操作符

publishOn 依赖 Worker 的串行性：如果 Worker 把任务并发跑在多个线程上，下游就会收到并发的 onNext，违反 Reactive Streams 的信号串行要求。自己实现 Worker 时，这一点必须用单线程执行器保证。

手写一个 LoggingScheduler

下面的实现只做立即调度（不实现延迟/周期），重点展示三条契约：直接调度可以并发、Worker 内严格串行、dispose() 之后拒绝新任务并回收线程。

import reactor.core.Disposable;
import reactor.core.scheduler.Scheduler;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public final class LoggingScheduler implements Scheduler {

    private final ExecutorService pool;
    private final AtomicInteger threadSeq = new AtomicInteger();
    private final AtomicInteger workerSeq = new AtomicInteger();
    private final Set<LoggingWorker> workers = ConcurrentHashMap.newKeySet();
    private final AtomicBoolean disposed = new AtomicBoolean();

    public LoggingScheduler(String name, int threads) {
        ThreadFactory factory = r ->
            new Thread(r, name + "-pool-" + threadSeq.incrementAndGet());
        this.pool = Executors.newFixedThreadPool(threads, factory);
    }

    @Override
    public Disposable schedule(Runnable task) {
        if (disposed.get()) {
            throw new RejectedExecutionException("scheduler disposed");
        }
        Future<?> future = pool.submit(() -> {
            System.out.println("[direct] on "
                + Thread.currentThread().getName());
            task.run();
        });
        return disposable(future);
    }

    @Override
    public Worker createWorker() {
        if (disposed.get()) {
            throw new RejectedExecutionException("scheduler disposed");
        }
        LoggingWorker worker = new LoggingWorker(workerSeq.incrementAndGet());
        workers.add(worker);
        return worker;
    }

    @Override
    public void dispose() {
        if (disposed.compareAndSet(false, true)) {
            for (LoggingWorker worker : workers) {
                worker.dispose();
            }
            workers.clear();
            pool.shutdownNow();
        }
    }

    @Override
    public boolean isDisposed() {
        return disposed.get();
    }

    private static Disposable disposable(Future<?> future) {
        return new Disposable() {
            @Override public void dispose() { future.cancel(true); }
            @Override public boolean isDisposed() { return future.isDone(); }
        };
    }

    private final class LoggingWorker implements Worker {

        private final int id;
        private final ScheduledThreadPoolExecutor executor;
        private final AtomicBoolean workerDisposed = new AtomicBoolean();

        LoggingWorker(int id) {
            this.id = id;
            this.executor = new ScheduledThreadPoolExecutor(1,
                r -> new Thread(r, "worker-" + id));
            this.executor.setRemoveOnCancelPolicy(true);
        }

        @Override
        public Disposable schedule(Runnable task) {
            if (workerDisposed.get()) {
                throw new RejectedExecutionException("worker disposed");
            }
            Future<?> future = executor.submit(() -> {
                System.out.println("[worker " + id + "] on "
                    + Thread.currentThread().getName());
                task.run();
            });
            return disposable(future);
        }

        @Override
        public void dispose() {
            if (workerDisposed.compareAndSet(false, true)) {
                executor.shutdownNow();
                workers.remove(this);
            }
        }

        @Override
        public boolean isDisposed() {
            return workerDisposed.get();
        }
    }
}

三个设计点：

  1. 直接调度用固定线程池：Scheduler.schedule 的契约本来就不保证顺序，两个任务落在 custom-pool-1、custom-pool-2 上都合法。
  2. 每个 Worker 一个单线程 ScheduledThreadPoolExecutor：保证同一 Worker 的任务串行且 FIFO。setRemoveOnCancelPolicy(true) 让被取消的排队任务及时移出队列，否则它会一直占着位置。
  3. workers 集合由 Scheduler 统一回收：Worker 线程默认是非守护线程，Scheduler.dispose() 如果不关闭它们，JVM 会一直挂住。这一步在实际调试中很容易漏掉。

运行验证

import reactor.core.scheduler.Scheduler;

public class CustomSchedulerDemo {

    public static void main(String[] args) throws Exception {
        LoggingScheduler scheduler = new LoggingScheduler("custom", 2);

        for (int i = 1; i <= 4; i++) {
            int id = i;
            scheduler.schedule(() -> {
                sleep(120);
                System.out.println("direct " + id + " done");
            });
        }

        Scheduler.Worker worker = scheduler.createWorker();
        for (int i = 1; i <= 4; i++) {
            int id = i;
            worker.schedule(() -> {
                sleep(60);
                System.out.println("worker task " + id + " done");
            });
        }

        Thread.sleep(500);

        Scheduler.Worker doomed = scheduler.createWorker();
        doomed.schedule(() -> {
            try {
                Thread.sleep(500);
                System.out.println("long task finished");
            } catch (InterruptedException e) {
                System.out.println("long task interrupted");
            }
        });
        doomed.schedule(() -> System.out.println("queued task ran"));
        Thread.sleep(50);
        doomed.dispose();

        try {
            doomed.schedule(() -> System.out.println("after dispose"));
        } catch (RuntimeException e) {
            System.out.println("after dispose: "
                + e.getClass().getSimpleName());
        }

        Thread.sleep(600);
        scheduler.dispose();
        System.out.println("scheduler disposed=" + scheduler.isDisposed());
    }

    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

一次实测输出：

[direct] on custom-pool-1
[direct] on custom-pool-2
[worker 1] on worker-1
worker task 1 done
[worker 1] on worker-1
direct 1 done
[direct] on custom-pool-1
direct 2 done
[direct] on custom-pool-2
worker task 2 done
[worker 1] on worker-1
worker task 3 done
[worker 1] on worker-1
direct 3 done
direct 4 done
worker task 4 done
[worker 2] on worker-2
long task interrupted
after dispose: RejectedExecutionException
scheduler disposed=true

对号入座：

  • direct 1..4 分布在 custom-pool-1 和 custom-pool-2，完成顺序由线程调度决定，多跑几次可能不同；
  • worker task 1..4 全部在 worker-1，且严格按提交顺序打印；
  • doomed.dispose() 中断了正在运行的 worker-2 任务，队列里的 queued task ran 没有执行；
  • 关闭后的 Worker 再 schedule 抛 RejectedExecutionException；
  • 最后只调用 scheduler.dispose()，没有单独调用 worker.dispose()，JVM 仍能正常退出，因为 Scheduler 回收了自己创建的 Worker。

虚拟时间：测试时间相关操作

VirtualTimeScheduler 在 reactor-test 里，包名是 reactor.test.scheduler，不是 reactor-core。最常见用法是交给 StepVerifier 注入：

import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

import java.time.Duration;

@Test
void intervalUsesVirtualClock() {
    StepVerifier.withVirtualTime(() ->
            Flux.interval(Duration.ofMinutes(1))
                .map(i -> "tick" + i)
                .take(3))
        .expectSubscription()
        .expectNoEvent(Duration.ofMinutes(1))
        .expectNext("tick0")
        .thenAwait(Duration.ofMinutes(1))
        .expectNext("tick1")
        .thenAwait(Duration.ofMinutes(1))
        .expectNext("tick2")
        .verifyComplete();
}

这段测试不需要真的等 3 分钟。关键是所有依赖时间的操作符都要在 lambda 内创建：withVirtualTime 会在验证期间替换 Schedulers 工厂，lambda 外创建的操作符拿不到注入的虚拟时钟。

一个常见的错误写法是先在外部创建调度器和流，再只把流交给 withVirtualTime：

VirtualTimeScheduler vts = VirtualTimeScheduler.create();
Flux<String> flux = Flux.interval(Duration.ofMinutes(1), vts)
    .map(i -> "tick" + i)
    .take(3);

// 错误：注入的是另一个 VirtualTimeScheduler，推进时钟不会影响 vts
StepVerifier.withVirtualTime(() -> flux)
    .expectSubscription()
    .thenAwait(Duration.ofMinutes(3))
    .expectNextCount(3)
    .verifyComplete();

实测这个写法 10 秒仍未返回：thenAwait 推进的是注入的调度器，而 flux 用的是外部那个 vts。如果确实要传入已有的调度器，必须用三参数重载，并保证两边是同一个实例：

StepVerifier.withVirtualTime(
        () -> Flux.interval(Duration.ofMinutes(1), vts).take(3),
        () -> vts,
        Long.MAX_VALUE)   // 第三个参数是初始 request，不能小于断言需要的元素数
    .expectSubscription()
    .thenAwait(Duration.ofMinutes(3))
    .expectNextCount(3)
    .verifyComplete();

这里第三个参数如果写成 1，expectNextCount(3) 会直接报错：

The scenario will hang at expectNextCount(3) due to too little request
being performed for the expectations to finish; request remaining since last step: 1, expected: 3

不用 StepVerifier 时，也可以手动推进时钟：

VirtualTimeScheduler scheduler = VirtualTimeScheduler.create();
AtomicReference<Long> value = new AtomicReference<>();
AtomicInteger completed = new AtomicInteger();

Mono.delay(Duration.ofSeconds(10), scheduler)
    .flatMap(i -> Mono.delay(Duration.ofSeconds(5), scheduler))
    .subscribe(value::set, error -> error.printStackTrace(),
        completed::incrementAndGet);

System.out.println("before: value=" + value.get()
    + " completed=" + completed.get());
scheduler.advanceTimeBy(Duration.ofSeconds(10));
System.out.println("+10s: value=" + value.get()
    + " completed=" + completed.get());
scheduler.advanceTimeBy(Duration.ofSeconds(5));
System.out.println("+15s: value=" + value.get()
    + " completed=" + completed.get());

实测输出：

before: value=null completed=0
+10s: value=null completed=0
+15s: value=0 completed=1

advanceTimeBy(10s) 只执行 10 秒内到期的任务：外层 Mono.delay(10s) 触发并订阅内层 Mono.delay(5s)，内层要到 t=15s 才到期，所以第一次推进后还没有值。第二次再推 5 秒，整条链才完成。VirtualTimeScheduler 还提供 advanceTimeTo(Instant) 和 getScheduledTaskCount()，后者可以断言「时钟推进后是否还有挂起的定时任务」。

按 key 路由到固定线程

分区调度器适合「同一个用户/订单的操作必须串行，不同 key 之间可以并行」的场景：

import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

public final class PartitionedScheduler {

    private final Scheduler[] partitions;
    private final int partitionCount;

    public PartitionedScheduler(int partitionCount, String name) {
        if (partitionCount <= 0) {
            throw new IllegalArgumentException("partitionCount must be > 0");
        }
        this.partitionCount = partitionCount;
        this.partitions = new Scheduler[partitionCount];
        for (int i = 0; i < partitionCount; i++) {
            partitions[i] = Schedulers.newSingle(name + "-" + i);
        }
    }

    public Scheduler forKey(String key) {
        int idx = Math.floorMod(key.hashCode(), partitionCount);
        return partitions[idx];
    }

    public void dispose() {
        for (Scheduler scheduler : partitions) {
            scheduler.dispose();
        }
    }
}

PartitionedScheduler ps = new PartitionedScheduler(4, "user");
for (String key : Arrays.asList(
        "user123", "user456", "user123", "user789", "user123")) {
    Mono.fromCallable(() -> {
        System.out.println(key + " on "
            + Thread.currentThread().getName());
        return key;
    }).subscribeOn(ps.forKey(key)).block();
}
ps.dispose();

实测输出：

user123 on user-3-1
user456 on user-2-2
user123 on user-3-1
user789 on user-1-3
user123 on user-3-1

同一个 user123 三次都落在 user-3-1；不同 key 分散到不同分区。线程名末尾的编号来自 Reactor 的全局计数器，分区编号在前缀里，别用末尾编号推断分区数。

几个约束：

  • Math.floorMod 处理负数 hash；直接 % partitionCount 会得到负下标。
  • 分区数变化会重新映射所有 key，运行中的流不能靠「改配置」重新平衡。
  • Schedulers.newSingle(name) 默认是非守护线程，dispose() 必须执行，否则 JVM 退不出去。
  • subscribeOn 只把订阅和请求放到分区线程上；如果内层操作符又用 publishOn 切走，串行性就不再由分区保证。
  • 同一个 key 的高并发请求会在该分区的单线程上排队，分区数太少会形成热点。

包装现有调度器：别忘了转发延迟与周期方法

如果只是给任务统一计时，优先写装饰器，而不是从零实现 Scheduler。但装饰器必须把三类 schedule 方法都转发出去：

import reactor.core.Disposable;
import reactor.core.scheduler.Scheduler;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

public final class TimingScheduler implements Scheduler {

    private final Scheduler delegate;
    private final LongAdder executions = new LongAdder();
    private final AtomicLong totalNanos = new AtomicLong();

    public TimingScheduler(Scheduler delegate) {
        this.delegate = delegate;
    }

    @Override
    public Disposable schedule(Runnable task) {
        return delegate.schedule(timed(task));
    }

    @Override
    public Disposable schedule(Runnable task, long delay, TimeUnit unit) {
        return delegate.schedule(timed(task), delay, unit);
    }

    @Override
    public Disposable schedulePeriodically(Runnable task,
                                           long initialDelay, long period,
                                           TimeUnit unit) {
        return delegate.schedulePeriodically(
            timed(task), initialDelay, period, unit);
    }

    @Override
    public Worker createWorker() {
        Worker worker = delegate.createWorker();
        return new Worker() {
            @Override
            public Disposable schedule(Runnable task) {
                return worker.schedule(timed(task));
            }

            @Override
            public Disposable schedule(Runnable task, long delay, TimeUnit unit) {
                return worker.schedule(timed(task), delay, unit);
            }

            @Override
            public Disposable schedulePeriodically(Runnable task,
                                                   long initialDelay, long period,
                                                   TimeUnit unit) {
                return worker.schedulePeriodically(
                    timed(task), initialDelay, period, unit);
            }

            @Override
            public void dispose() {
                worker.dispose();
            }

            @Override
            public boolean isDisposed() {
                return worker.isDisposed();
            }
        };
    }

    @Override
    public long now(TimeUnit unit) {
        return delegate.now(unit);
    }

    @Override
    public void dispose() {
        delegate.dispose();
    }

    @Override
    public boolean isDisposed() {
        return delegate.isDisposed();
    }

    private Runnable timed(Runnable task) {
        return () -> {
            long start = System.nanoTime();
            try {
                task.run();
            } finally {
                totalNanos.addAndGet(System.nanoTime() - start);
                executions.increment();
            }
        };
    }

    public long executions() {
        return executions.sum();
    }

    public double averageMillis() {
        long count = executions.sum();
        return count == 0 ? 0 : totalNanos.get() / 1_000_000.0 / count;
    }
}

用自建的 newParallel 验证（不要包装全局 Schedulers.parallel() 后再调用 dispose()，那会关掉全进程共享的池）：

TimingScheduler timing =
    new TimingScheduler(Schedulers.newParallel("timed", 2));

Flux.interval(Duration.ofMillis(50), timing)
    .take(4)
    .blockLast();

Mono.fromCallable(() -> "ok").subscribeOn(timing).block();

System.out.println("executions=" + timing.executions());
timing.dispose();

实测 executions=5：4 次 interval tick 加 1 次 subscribeOn 的订阅任务。averageMillis() 的值随机器和任务内容变化，只适合做趋势观察，不适合写精确断言。

如果漏掉 schedulePeriodically 的转发，Flux.interval(..., timing) 会立刻失败：

reactor.core.Exceptions$ReactorRejectedExecutionException: Scheduler unavailable

因为 Worker/Scheduler 的默认周期实现本身就会抛「not time capable」，包装层不转发就等于把它关掉了。

常见错误

错误后果正确做法
只实现 schedule、createWorker编译失败，缺 isDisposed()Scheduler 继承 Disposable，两个方法都要实现
Worker 共用 Scheduler 的线程池同一 Worker 的任务可能并发，破坏 publishOn 的串行假设每个 Worker 用独立单线程执行器
Scheduler.dispose() 不关闭 Worker非守护线程泄漏，JVM 不退出在 Scheduler 内登记并回收所有 Worker
认为 Scheduler.schedule 有序任务完成顺序与提交顺序不一致需要顺序就用同一个 Worker，或 concatMap
自定义调度器不实现延迟/周期interval/delayElements 报 Scheduler unavailable转发到支持定时的执行器，或明确只用立即调度
包装 fromExecutor(executor) 的 Worker接口不保证 FIFO 和串行用 fromExecutor(executor, true) 开启 trampoline，或 newSingle
包装全局调度器并转发 dispose()关掉全进程共享的 parallel/single明确所有权：借用的调度器不转发 dispose

小结

关注点结论
是否要手写优先用 newSingle/newParallel/newBoundedElastic/fromExecutorService
必须实现schedule(Runnable)、createWorker()、isDisposed()
延迟/周期默认抛 RejectedExecutionException，需要时显式转发
Worker 串行由实现保证，单线程执行器是最直接的做法
生命周期Scheduler.dispose() 负责它创建的 Worker 与线程池
测试StepVerifier.withVirtualTime 或手动 advanceTimeBy
分区newSingle + floorMod(key.hashCode(), n)，并处理热点
装饰器三类 schedule 方法都要转发，明确是否拥有 delegate

下一章进入背压策略——响应式编程最核心的生产问题。

阅读导航

上一篇：6.2 BoundedElastic 与线程池模型

下一篇：7.1 背压策略：Buffer

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南