java

12.1 生命周期钩子：doOn* 系列

用实测输出纠正几个常见误解：doOnTerminate 在下游收到 onComplete/onError 之前执行，doAfterTerminate 在下游处理完之后执行；doOnEach 只看到 onNext/onError/onComplete，看不到 onSubscribe 和 cancel；doOnRequest 报告的是请求量而不是实际产量。同时给出 doFinally 的 SIGNAL 类型、取消路径的顺序、钩子所在线程，以及为什么不能用一次 nanoTime 测量钩子开销。

Leeting Yan 5 分钟阅读 2093 字

钩子清单

Reactor 的 doOn* 钩子在信号经过时插入副作用，不改变元素本身：

钩子触发时机能否区分终止类型
doFirst订阅开始、向上游传播之前—
doOnSubscribe收到 onSubscribe—
doOnRequest下游发出 request(n)—
doOnNext每个 onNext—
doOnErroronError只有错误
doOnCompleteonComplete只有完成
doOnCancel下游取消只有取消
doOnTerminate终止信号向下游传播之前否
doAfterTerminate终止信号被下游处理之后否
doFinally终止或取消之后是，收到 SignalType
doOnDiscard元素被操作符丢弃—
doOnEach每个 onNext/onError/onComplete是，收到 Signal

doOnTerminate 和 doAfterTerminate 都不带参数，只能执行同一个动作；需要知道是完成、错误还是取消，用 doFinally。

完整链路的真实顺序

把六个钩子放在同一条链上，下游用 BaseSubscriber 显式请求 3 个元素：

Flux.range(1, 3)
    .doOnSubscribe(s -> System.out.println("doOnSubscribe"))
    .doOnRequest(n -> System.out.println("doOnRequest(" + n + ")"))
    .doOnNext(i -> System.out.println("doOnNext " + i))
    .doOnComplete(() -> System.out.println("doOnComplete"))
    .doOnTerminate(() -> System.out.println("doOnTerminate"))
    .doAfterTerminate(() -> System.out.println("doAfterTerminate"))
    .doFinally(st -> System.out.println("doFinally " + st))
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription s) {
            System.out.println("subscriber.onSubscribe");
            request(3);
        }
        @Override
        protected void hookOnNext(Integer v) {
            System.out.println("subscriber.onNext " + v);
        }
        @Override
        protected void hookOnComplete() {
            System.out.println("subscriber.onComplete");
        }
    });

实测输出（Reactor 3.8.8，JDK 25）：

doOnSubscribe
subscriber.onSubscribe
doOnRequest(3)
doOnNext 1
subscriber.onNext 1
doOnNext 2
subscriber.onNext 2
doOnNext 3
subscriber.onNext 3
doOnComplete
doOnTerminate
subscriber.onComplete
doFinally onComplete
doAfterTerminate

三个容易看错的地方：

  1. doOnTerminate 在下游的 onComplete 之前。它拿到的是「即将终止」的通知，不是「下游已经终止」的通知。
  2. doAfterTerminate 在下游处理完之后。它才适合做「下游已经收尾」的清理。
  3. doFinally 在两者之间。它在下游 onComplete 返回后执行，但又位于 doAfterTerminate 的下游，所以先打印。

doFinally 与 doAfterTerminate 的相对顺序取决于它们在链中的位置；不要依赖固定顺序，只依赖各自「在终止前/后」的语义。

错误路径

Flux.just(1, 2, 0)
    .map(i -> 10 / i)
    .doOnError(e -> System.out.println("doOnError " + e))
    .doOnTerminate(() -> System.out.println("doOnTerminate"))
    .doAfterTerminate(() -> System.out.println("doAfterTerminate"))
    .doFinally(st -> System.out.println("doFinally " + st))
    .subscribe(v -> System.out.println("subscriber.onNext " + v),
               e -> System.out.println("subscriber.onError " + e));

实测：

subscriber.onNext 10
subscriber.onNext 5
doOnError java.lang.ArithmeticException: / by zero
doOnTerminate
subscriber.onError java.lang.ArithmeticException: / by zero
doFinally onError
doAfterTerminate

doOnError 先于 doOnTerminate，两者都在下游收到 onError 之前。下游的 onError 返回后，才轮到 doFinally 和 doAfterTerminate。

取消路径

doOnCancel 只在取消时触发，完成和错误都不会触发它。doFinally 则三种情况都会触发：

Disposable d = Flux.interval(Duration.ofMillis(50))
    .doOnCancel(() -> System.out.println("doOnCancel"))
    .doFinally(st -> System.out.println("doFinally " + st))
    .subscribe(i -> System.out.println("next " + i));

Thread.sleep(120);
d.dispose();

实测：

next 0
next 1
doOnCancel
doFinally cancel
disposed

如果要释放资源，doFinally 比 doOnCancel 更完整，因为错误和正常完成也会走它。更严格的资源管理用 Flux.using(...)，它把「创建—使用—释放」绑定在一个生命周期里，不依赖回调是否被调用。

doOnEach 看不到订阅和取消

很多文章说 doOnEach 覆盖「所有信号」，包括 onSubscribe。实测并非如此：

Flux.range(1, 2)
    .doOnEach(sig -> System.out.println(
        "doOnEach type=" + sig.getType()
        + " value=" + (sig.isOnNext() ? sig.get() : "-")))
    .subscribe();

输出：

doOnEach type=onNext value=1
doOnEach type=onNext value=2
doOnEach type=onComplete value=-

没有 onSubscribe，也没有 cancel。Signal 类型确实定义了订阅信号，但 doOnEach 的分发点只覆盖三个终止/元素信号。需要订阅用 doOnSubscribe，需要取消用 doOnCancel 或 doFinally。

这也意味着「用 doOnEach 打印完整生命周期」是不成立的：它会漏掉订阅和取消，而这两件事恰恰是排查泄漏时最需要的。

doOnRequest 报告请求量，不是产量

Flux.range(1, 10)
    .doOnRequest(n -> System.out.println("Request: " + n))
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription s) {
            request(3);
        }
        @Override
        protected void hookOnNext(Integer v) {
            System.out.println("got " + v);
            if (v % 3 == 0) {
                request(3);
            }
        }
    });

实测：

Request: 3
got 1
got 2
got 3
Request: 3
got 4
got 5
got 6
Request: 3
got 7
got 8
got 9
Request: 3
got 10

最后一行是 Request: 3，不是 Request: 1。钩子记录的是下游请求了多少，不是上游实际还能生产多少。要统计实际产量，应该数 doOnNext 的次数，而不是把 doOnRequest 的参数相加。

另一个细节：range 在发完 10 个元素后直接完成，并没有为了满足最后一次 request(3) 再要 2 个不存在的元素。请求量可以大于剩余元素数，这不是错误。

钩子在哪条线程执行

Flux.range(1, 3)
    .publishOn(Schedulers.boundedElastic())
    .doOnNext(i -> System.out.println("doOnNext " + i + " on " + Thread.currentThread().getName()))
    .map(i -> i * 2)
    .subscribeOn(Schedulers.parallel())
    .subscribe(i -> System.out.println("subscribe " + i + " on " + Thread.currentThread().getName()));

Thread.sleep(500);

实测：

doOnNext 1 on boundedElastic-1
subscribe 2 on boundedElastic-1
doOnNext 2 on boundedElastic-1
subscribe 4 on boundedElastic-1
doOnNext 3 on boundedElastic-1
subscribe 6 on boundedElastic-1

doOnNext 位于 publishOn 之后，所以在 boundedElastic 线程执行；下游的 subscribe 也在同一线程。注意第一对是 doOnNext 1 和 subscribe 2：doOnNext 在 map 之前看到原始值，订阅者看到的是翻倍后的值。

subscribeOn 控制的是订阅动作在哪个线程发生，不决定元素交付线程；这里元素交付由 publishOn 决定。所以订阅者跑在 boundedElastic，而不是 parallel。doOnSubscribe 能看到订阅线程，doOnNext 看到的是交付线程，两者可以不同。

生产使用的边界

  • 不要在钩子里改元素。doOnNext 的返回值是 void，看起来改不了，但消费可变对象仍然会改变下游看到的内容。转换用 map，副作用留在钩子。
  • 不要让钩子抛异常。doOnNext 抛出的异常会把正常流变成错误流；doOnError 里再抛会替换原异常。日志框架自身失败要吞掉。
  • 不要在高频路径里做同步 I/O。doOnNext(System.out::println) 的真正成本是控制台输出，不是钩子本身。
  • 指标用原子计数器或 Micrometer，不要在钩子里 synchronized。
  • doOnDiscard 用于丢弃路径。filter、take、溢出策略等会丢弃元素，需要释放池化对象时在这里回收。

关于「钩子带来多少开销」，不要用一次 System.nanoTime() 的差值下结论。JIT 预热、死代码消除、GC 都会让结果偏离；doOnNext(i -> {}) 这种空 lambda 甚至可能被优化掉。要用 JMH，固定 JDK、GC 和数据规模，报告分位数，而不是一个平均值。

小结

问题结论
doOnTerminate 何时执行下游收到终止信号之前
doAfterTerminate 何时执行下游处理完终止信号之后
doFinally 能区分什么onComplete、onError、cancel
doOnEach 覆盖范围元素、错误、完成；不含订阅和取消
doOnRequest 的参数下游请求量，不是实际产量
取消时触发谁doOnCancel、doFinally
钩子线程钩子所在操作符的线程，由上游调度器决定
性能用 JMH 测，不要用一次性计时

阅读导航

上一篇：11.3 Subscriber 与 Processor 合规测试

下一篇：12.2 调试技术：Hooks 与检查点

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

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