java

12.2 调试技术：Hooks 与检查点

用同一段代码的实测栈对比三种手段：默认 subscribe() 会把错误交给 onErrorDropped 并只打印操作符内部帧；Hooks.onOperatorDebug() 增加 Assembly trace 与 Error has been observed 站点；checkpoint() 只在标记处报告，且 3.8.8 只有三个重载，没有所谓「条件检查点」。同时给出 log() 的真实输出格式、MDC 跨线程失效的原因，以及为什么不能用一次计时估算调试开销。

Leeting Yan 5 分钟阅读 2234 字

先看默认行为：错误可能被丢弃

很多调试文章直接从「异常栈只有操作符内部帧」讲起，但更基础的问题是：subscribe() 不传错误回调时，错误根本不会抛给调用方。

import reactor.core.publisher.Flux;

public class DropDemo {
    public static void main(String[] args) {
        Flux.just(1, 2, 3)
            .map(i -> {
                if (i == 2) throw new RuntimeException("boom at " + i);
                return i * 2;
            })
            .subscribe();
    }
}

接上 SLF4J 后端后实测：

[main] ERROR reactor.core.publisher.Operators - Operator called default onErrorDropped
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.RuntimeException: boom at 2
Caused by: java.lang.RuntimeException: boom at 2
	at DropDemo.lambda$main$0(DropDemo.java:7)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:114)
	...

如果没有日志后端，这段输出是空的。错误被 onErrorDropped 记录后丢弃，程序继续运行。生产里最危险的情况不是栈不好看，而是异常完全消失。

所以调试的第一步是给 subscribe 传错误回调，或者用 blockLast() 让异常回到调用线程：

.subscribe(v -> { }, e -> e.printStackTrace());

默认栈能看到什么

同一段代码，只把 subscribe() 换成带错误回调的版本：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;

public class DebugDemo {
    public static void main(String[] args) {
        // Hooks.onOperatorDebug();
        Flux.just(1, 2, 3)
            .map(i -> {
                if (i == 2) {
                    throw new RuntimeException("boom at " + i);
                }
                return i * 2;
            })
            .subscribe(v -> { }, e -> e.printStackTrace());
    }
}

实测：

java.lang.RuntimeException: boom at 2
	at DebugDemo.lambda$main$0(DebugDemo.java:10)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:114)
	at reactor.core.publisher.FluxArray$ArraySubscription.fastPath(FluxArray.java:171)
	at reactor.core.publisher.FluxArray$ArraySubscription.request(FluxArray.java:96)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:172)
	at reactor.core.publisher.LambdaSubscriber.onSubscribe(LambdaSubscriber.java:123)
	...
	at DebugDemo.main(DebugDemo.java:16)

第一帧能定位到 lambda（第 10 行），但看不到这个 lambda 属于哪条链路。栈里只有 FluxMapFuseable、FluxArray 这些操作符内部类；当项目里有十几条相似的流时，仅凭这些帧无法判断是哪一个 map。

Hooks.onOperatorDebug()

把注释打开，并且必须在装配之前调用：

public static void main(String[] args) {
    Hooks.onOperatorDebug();
    Flux.just(1, 2, 3)
        // ...
}

实测输出：

java.lang.RuntimeException: boom at 2
	at DebugDemo.lambda$main$0(DebugDemo.java:10)
	Suppressed: The stacktrace has been enhanced by Reactor, refer to additional information below: 
Assembly trace from producer [reactor.core.publisher.FluxMapFuseable] :
	reactor.core.publisher.Flux.map(Flux.java:6628)
	DebugDemo.main(DebugDemo.java:8)
Error has been observed at the following site(s):
	*__Flux.map ⇢ at DebugDemo.main(DebugDemo.java:8)
Original Stack Trace:
		at DebugDemo.lambda$main$0(DebugDemo.java:10)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:114)
		...

多了三段信息：

段含义
Assembly trace from producer这个操作符在哪个类的哪一行被创建
Error has been observed at错误经过的站点，*__ 标出错处
Original Stack Trace原始栈，保留操作符内部帧

Assembly trace 里的 DebugDemo.java:8 是 .map(...) 所在行，不是 Flux.just 的行。Reactor 只记录出错操作符的装配位置；不要期待它把整条链的每一行都列出来。

全局开关的代价

Hooks.onOperatorDebug() 影响的是整个 JVM：开启之后组装的每个操作符都会额外包装一层 FluxOnAssembly，并在装配时抓取调用栈。代价包括：

  • 装配阶段的栈抓取；
  • 每个操作符多一个包装对象；
  • 异常对象上挂载的装配快照。

具体慢多少取决于操作符数量、链路深度和是否命中 JIT，不能用一个固定的百分比概括。要量化就用 JMH 对比同一条链在开关前后的吞吐与分配率。

它只适合本地开发和测试。线上需要时，优先用 checkpoint()；如果确实要在线开启，至少用配置中心控制，并在停止排查后调用：

Hooks.resetOnOperatorDebug();

checkpoint()：只在关键位置标记

checkpoint() 不改变数据流，只在装配时给当前链打一个标记。错误发生并经过这个标记时，报告里会出现站点：

import reactor.core.publisher.Flux;

public class CheckpointDemo {
    public static void main(String[] args) {
        Flux.just(1, 2, 3)
            .map(i -> {
                if (i == 2) throw new RuntimeException("boom");
                return i;
            })
            .checkpoint("after map")
            .filter(i -> i > 2)
            .subscribe(v -> { }, e -> e.printStackTrace());
    }
}

实测：

java.lang.RuntimeException: boom
	at CheckpointDemo.lambda$main$0(CheckpointDemo.java:7)
	Suppressed: The stacktrace has been enhanced by Reactor, refer to additional information below: 
Error has been observed at the following site(s):
	*__checkpoint ⇢ after map
Original Stack Trace:
		at CheckpointDemo.lambda$main$0(CheckpointDemo.java:7)
		at reactor.core.publisher.FluxMapFuseable$MapFuseableConditionalSubscriber.onNext(FluxMapFuseable.java:284)
		...

注意两点：

  1. 输出里是 *__checkpoint ⇢ after map，没有方法名和括号。
  2. checkpoint 报告的是错误经过的最后一个检查点。如果一条链上有多个 checkpoint，只有离错误最近的那个会出现在站点列表里，其他标记不会全部列出。

3.8.8 只有三个重载

flux.checkpoint();
flux.checkpoint("after map");
flux.checkpoint("after map", true);

第三个参数是 forceStackTrace，不是「条件模式」。Reactor 3.8.8 的 Flux 上没有 checkpoint(String, Predicate) 之类的重载，也不存在「只在错误时记录」的检查点。所谓「条件检查点」是误传。

checkpoint() 适合在关键边界打点，例如数据库查询之后、外部调用返回之后、flatMap 汇合之前。它比全局 hook 便宜，但仍然有装配开销；不要在每个操作符后面都加一个。

log()：查看信号流

log() 把 onSubscribe、request、onNext、onError、onComplete、cancel 打印到 SLF4J。它需要日志后端，否则什么都看不到。

Flux.range(1, 3)
    .log("myStream")
    .map(i -> i * 2)
    .subscribe();

接 slf4j-simple 后实测：

[main] INFO myStream - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
[main] INFO myStream - | request(unbounded)
[main] INFO myStream - | onNext(1)
[main] INFO myStream - | onNext(2)
[main] INFO myStream - | onNext(3)
[main] INFO myStream - | onComplete()

前缀 [main] INFO myStream - 是 slf4j-simple 的格式；换 Logback 后格式会变，但 | onSubscribe(...) 这部分是 Reactor 输出的。log() 适合确认「谁请求了多少、收到了什么」，不适合长期留在生产：每个信号一条日志，高频流会把磁盘打满。

生产环境：别把 MDC 当万能追踪

一个常见的错误写法：

Flux.just(1, 2, 3)
    .doOnNext(i -> MDC.put("element", String.valueOf(i)))
    .publishOn(Schedulers.boundedElastic())
    .doOnError(e -> log.error("Failed at element {}", MDC.get("element"), e))
    .subscribe();

MDC 是 ThreadLocal。publishOn 会把后续信号切到另一条线程；在 doOnNext 里 put 的值，到了 doOnError 所在的线程通常是空的。即使没有 publishOn，异步源也可能在不同线程发信号。

可行的做法有两类：

1. 在同一个回调里完成 put、日志、remove：

.doOnNext(v -> {
    MDC.put("element", String.valueOf(v));
    try {
        log.info("processing");
    } finally {
        MDC.remove("element");
    }
})

这能保证日志和 MDC 在同一线程，但只覆盖当前这一条日志。

2. 用 Reactor Context 做跨操作符传递。 把 traceId 写入 contextWrite，在需要的地方用 deferContextual 读取；再配合 Micrometer 的 Context Propagation 让 MDC 自动跟随 Reactor Context。代价是引入额外依赖，并且要开启自动传播，不是默认行为。

如果只是排查一条具体链路，checkpoint 加结构化日志通常比全局 MDC 更直接。

调试顺序建议

  1. 先确认错误没有被丢弃：subscribe(v -> {}, e -> ...) 或 blockLast()。
  2. 需要定位装配位置：在可疑边界加 checkpoint("...")。
  3. checkpoint 不够、需要整条链：本地开启 Hooks.onOperatorDebug()，复现后立即关闭。
  4. 需要看请求与信号：临时加 log("name")，确认背压行为后移除。
  5. 线上：结构化日志 + traceId，不长期开启全局 hook。

小结

工具得到什么主要限制
默认栈出错 lambda 与操作符内部帧不知道属于哪条链；无错误回调时错误被丢弃
onOperatorDebug()装配位置、错误站点、原始栈全局生效，装配开销大
checkpoint()标记处的错误站点只有三个重载，无「条件模式」
log()完整信号流需要日志后端，输出量大
Context + Micrometer跨线程 traceId需要额外依赖与显式开启

阅读导航

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

下一篇：12.3 装配时 vs 订阅时

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

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