调试 Reactor
目录
从命令式、同步的编程范式切换到响应式、异步的范式，有时会令人望而生畏。学习曲线中最陡峭的环节之一，就是当出现问题时如何分析和调试。
在命令式的世界里，调试通常相当直接。您可以阅读堆栈跟踪，看问题源自何处。完全是您自己代码的失败吗？失败发生在某个库代码中吗？如果是，您代码的哪一部分调用了该库，可能传入了不当参数，最终导致了失败？
1. 典型的 Reactor 堆栈跟踪
当您转向异步代码时，事情可能会变得复杂得多。
考虑下面的堆栈跟踪：
java.lang.IndexOutOfBoundsException: Source emitted more than one item
at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.tryEmitScalar(FluxFlatMap.java:445)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:379)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
at reactor.core.publisher.FluxRange$RangeSubscription.slowPath(FluxRange.java:154)
at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:109)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.onSubscribe(FluxFlatMap.java:332)
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68)
at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63)
at reactor.core.publisher.FluxFlatMap.subscribe(FluxFlatMap.java:97)
at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58)
at reactor.core.publisher.Mono.subscribe(Mono.java:3096)
at reactor.core.publisher.Mono.subscribeWith(Mono.java:3204)
at reactor.core.publisher.Mono.subscribe(Mono.java:3090)
at reactor.core.publisher.Mono.subscribe(Mono.java:3057)
at reactor.core.publisher.Mono.subscribe(Mono.java:3029)
at reactor.guide.GuideTests.debuggingCommonStacktrace(GuideTests.java:995)
这里面牵扯到很多东西。我们得到一个 IndexOutOfBoundsException，它告诉我们 source emitted more than one item（源发出了多个元素）。
我们大概很快就会假设这个源是一个 Flux 或 Mono，下一行提到的 MonoSingle 证实了这一点。所以看起来像是 single 操作符在抱怨。
查阅 Mono#single 操作符的 Javadoc，我们看到 single 有一个约定：源必须恰好发出一个元素。看来我们的源发出了多于一个元素，从而违反了该约定。
我们能否深入挖掘并定位那个源呢？接下来的几行帮助不大。它们带我们穿过看似响应式链的内部，经过对 subscribe 和 request 的多次调用。
浏览这些行，我们至少可以开始勾勒出出错链的大致轮廓：它似乎涉及 MonoSingle、FluxFlatMap 和 FluxRange（每一个在跟踪中占了几行，但总体上就是这三个类参与其中）。所以也许是一个 range().flatMap().single() 链？
但如果我们在应用中大量使用这种模式呢？这仍然告诉不了我们太多，单纯搜索 single 也找不到问题所在。然后最后一行指向了我们自己的一些代码。终于，我们接近目标了。
不过等一下。当我们打开源文件时，看到的只是订阅了一个已存在的 Flux，如下所示：
toDebug
.subscribeOn(Schedulers.immediate())
.subscribe(System.out::println, Throwable::printStackTrace);
所有这些都发生在订阅时，但 Flux 本身并不是在这里声明的。更糟的是，当我们找到该变量声明的地方，看到的是：
public Mono<String> toDebug; //please overlook the public class attribute
变量并不是在声明处实例化的。我们必须设想最坏的情况：发现在应用中可能有好几条不同的代码路径对它赋值。我们仍然无法确定是哪一条导致了问题。
这有点类似于 Reactor 中运行时错误（相对于编译错误而言）的等价物。
我们希望更容易地找到的是，这个操作符是在何处被加入链中的——也就是说，Flux 是在何处声明的。我们通常称之为 Flux 的「装配」（assembly）。
2. 激活调试模式（又称 traceback）
本节描述的是启用调试能力最简单但也最慢的方式，因为它会为每个操作符捕获堆栈跟踪。有关更细粒度的调试方式，请参阅 checkpoint() 替代方案；有关更高级、性能更好的全局选项，请参阅 生产环境可用的全局调试。
尽管对于有一定经验的人来说，堆栈跟踪仍然能够传达一些信息，但我们可以看到，在更复杂的情况下，它本身并不理想。
幸运的是，Reactor 内置了专为调试而设计的装配期插桩（assembly-time instrumentation）。
做法是在应用启动时（或至少在出问题的 Flux 或 Mono 被实例化之前）通过 Hooks.onOperatorDebug() 方法激活全局调试模式，如下所示：
Hooks.onOperatorDebug();
这会对 Reactor 操作符方法的调用（即它们被装配进链的地方）进行插桩，方式是包装操作符的构造过程并在那里捕获堆栈跟踪。由于这是在声明操作符链时完成的，该钩子应在此之前激活，因此最稳妥的做法是在应用一开始就激活它。
之后，如果发生异常，出错的操作符就能引用这次捕获，并重新加工堆栈跟踪，附加额外信息。
我们把这种捕获到的装配信息（以及 Reactor 一般附加到异常上的额外信息）称为 traceback。
下一节中，我们会看到堆栈跟踪有何不同，以及如何解读这些新信息。
3. 在调试模式下阅读堆栈跟踪
当我们复用最初的示例，但激活 operatorStacktrace 调试特性时，会发生以下几件事：
-
指向订阅点、因而不太重要的堆栈跟踪，在第一帧之后被截断并放到一边。
-
一个特殊的 suppressed 异常被添加到原始异常上（如果已有则加以补充）。
-
为该特殊异常构造一条包含若干段落的消息。
-
第一段会回溯到出错操作符的装配点。
-
第二段会尝试展示由该操作符构建、并观测到错误传播的链。
-
最后一段是原始堆栈跟踪
完整的堆栈跟踪打印出来后如下：
java.lang.IndexOutOfBoundsException: Source emitted more than one item
at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:127) 1
Suppressed: The stacktrace has been enhanced by Reactor, refer to additional information below: 2
Assembly trace from producer [reactor.core.publisher.MonoSingle] : 3
reactor.core.publisher.Flux.single(Flux.java:7915)
reactor.guide.GuideTests.scatterAndGather(GuideTests.java:1017)
Error has been observed at the following site(s): 4
*_______Flux.single ⇢ at reactor.guide.GuideTests.scatterAndGather(GuideTests.java:1017) 5
|_ Mono.subscribeOn ⇢ at reactor.guide.GuideTests.debuggingActivated(GuideTests.java:1071) 6
Original Stack Trace: 7
at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:127)
...
8
...
at reactor.core.publisher.Mono.subscribeWith(Mono.java:4363)
at reactor.core.publisher.Mono.subscribe(Mono.java:4223)
at reactor.core.publisher.Mono.subscribe(Mono.java:4159)
at reactor.core.publisher.Mono.subscribe(Mono.java:4131)
at reactor.guide.GuideTests.debuggingActivated(GuideTests.java:1067)
原始堆栈跟踪被截断为单帧。
这是新增的：我们看到捕获堆栈的包装操作符。traceback 就是从这里开始出现的。
首先，我们得到操作符在何处装配的一些细节。
其次，我们得到错误沿其传播的操作符链的概念，按从第一到最后（错误点到订阅点）的顺序排列。
每个观测到错误的操作符都会被列出，并附带使用它的用户类和所在行。这里我们有一个「根」。
这里我们看到链的一个简单部分。
堆栈跟踪的其余部分被移到末尾……
……展示操作符内部的一些细节（因此我们在这里删去了片段的一部分）。
捕获到的堆栈跟踪作为一个 suppressed OnAssemblyException 附加到原始错误上。它有三个部分，但第一段最值得关注。它展示了触发异常的操作符的构造路径。这里，它表明导致问题的 single 实际上是在 scatterAndGather 方法中创建的。
既然我们已经掌握了足够的信息来找出罪魁祸首，就可以有针对性地查看那个 scatterAndGather 方法了：
private Mono<String> scatterAndGather(Flux<String> urls) {
return urls.flatMap(url -> doRequest(url))
.single(); 1
}
果然，我们的
single就在这里。
现在我们可以看到，错误的根源是一个 flatMap，它对几个 URL 执行多次 HTTP 调用，却又链式接上了过于严格的 single。经过短暂的 git blame 并与那行代码的作者快速沟通后，我们发现他本意是用限制更宽松的 take(1)。
我们的问题解决了。
现在看堆栈跟踪中的以下部分：
Error has been observed at the following site(s):
在这个特定示例中，traceback 的第二部分未必有趣，因为错误实际上发生在链的最后一个操作符（最接近 subscribe 的那个）。考虑另一个示例可能会更清楚：
FakeRepository.findAllUserByName(Flux.just("pedro", "simon", "stephane"))
.transform(FakeUtils1.applyFilters)
.transform(FakeUtils2.enrichUser)
.blockLast();
现在想象一下，在 findAllUserByName 内部，有一个 map 失败了。这时，我们会在 traceback 的第二部分看到以下内容：
Error has been observed at the following site(s):
*________Flux.map ⇢ at reactor.guide.FakeRepository.findAllUserByName(FakeRepository.java:27)
|_ Flux.map ⇢ at reactor.guide.FakeRepository.findAllUserByName(FakeRepository.java:28)
|_ Flux.filter ⇢ at reactor.guide.FakeUtils1.lambda$static$1(FakeUtils1.java:29)
|_ Flux.transform ⇢ at reactor.guide.GuideDebuggingExtraTests.debuggingActivatedWithDeepTraceback(GuideDebuggingExtraTests.java:39)
|_ Flux.elapsed ⇢ at reactor.guide.FakeUtils2.lambda$static$0(FakeUtils2.java:30)
|_ Flux.transform ⇢ at reactor.guide.GuideDebuggingExtraTests.debuggingActivatedWithDeepTraceback(GuideDebuggingExtraTests.java:40)
这对应于操作符链中收到错误通知的那一段：
-
异常源自第一个
map。它通过*连接符以及使用_作为缩进而被标识为根。 -
异常被第二个
map观测到（两者实际上都对应于findAllUserByName方法）。 -
然后它被一个
filter和一个transform观测到，这表明链的这一部分是由一个可复用的转换函数（这里是applyFilters工具方法）构建的。 -
最后，它被一个
elapsed和一个transform观测到。同样，elapsed是由第二个 transform 的转换函数应用的。
在某些情况下，同一个异常会沿多条链传播，这时「根」标记 *_ 能让我们更好地把这些链区分开。如果某个点被观测到多次，调用点信息之后会出现 (observed x times)。
例如，考虑下面的片段：
public class MyClass {
public void myMethod() {
Flux<String> source = Flux.error(sharedError);
Flux<String> chain1 = source.map(String::toLowerCase).filter(s -> s.length() < 4);
Flux<String> chain2 = source.filter(s -> s.length() > 5).distinct();
Mono<Void> when = Mono.when(chain1, chain2);
}
}
在上面的代码中，错误传播到 when，经过两条独立的链 chain1 和 chain2。这会产生包含以下内容的 traceback：
Error has been observed at the following site(s):
*_____Flux.error ⇢ at myClass.myMethod(MyClass.java:3) (observed 2 times)
|_ Flux.map ⇢ at myClass.myMethod(MyClass.java:4)
|_ Flux.filter ⇢ at myClass.myMethod(MyClass.java:4)
*_____Flux.error ⇢ at myClass.myMethod(MyClass.java:3) (observed 2 times)
|_ Flux.filter ⇢ at myClass.myMethod(MyClass.java:5)
|_ Flux.distinct ⇢ at myClass.myMethod(MyClass.java:5)
*______Mono.when ⇢ at myClass.myMethod(MyClass.java:7)
我们可以看到：
-
有 3 个「根」元素（
when才是真正的根）。 -
可以看到两条从
Flux.error出发的链。 -
两条链似乎都基于同一个
Flux.error源（observed 2 times）。 -
第一条链是
Flux.error().map().filter -
第二条链是
Flux.error().filter().distinct()
关于 traceback 与 suppressed 异常的说明：由于 traceback 是作为 suppressed 异常附加到原始错误上的，这在某种程度上会干扰另一种使用该机制的异常：复合异常（composite exception）。这类异常可以通过 Exceptions.multiple(Throwable…) 直接创建，也可以由某些可能合并多个出错源的操作符（如 Flux#flatMapDelayError）产生。它们可以通过 Exceptions.unwrapMultiple(Throwable) 解包成一个 List，此时 traceback 会被视为复合异常的一个组成部分，成为返回的 List 的一部分。如果这不理想，可以借助 Exceptions.isTraceback(Throwable) 检查来识别 traceback，并改用 Exceptions.unwrapMultipleExcludingTracebacks(Throwable) 将其排除在解包之外。
静态异常：优化错误处理时，有时会通过预先创建全局静态实例来复用异常，而不是在异常状态发生时再创建新实例。对于这类被复用的异常，traceback 会作为附加的 suppressed 异常不断累积，最终不可避免地导致内存泄漏。在大多数情况下，静态异常会与禁用堆栈跟踪捕获配合使用——后者才是创建异常开销的最大来源。然而，异常可以在不带堆栈跟踪的情况下动态创建，只需付出内存分配的边际成本，而无需静态实例。如果您的代码仍需依赖异常的静态实例，避免 traceback 造成内存泄漏的方法是禁用 suppression，代价是当这些异常沿响应式链向下传播时，将无法看到 traceback。
这里我们涉及的是一种插桩，而创建堆栈跟踪是有开销的。这就是为什么这个调试特性应当只在受控的方式下、作为最后手段来启用。
3.1. checkpoint() 替代方案
调试模式是全局的，会影响应用中装配进 Flux 或 Mono 的每一个操作符。这样做的好处是支持事后调试：无论出现什么错误，我们都能获得额外信息来调试它。
如前所述，这种全局认知是有代价的：会对性能产生影响（因为填充了大量堆栈跟踪）。如果我们对可能出问题的操作符有所了解，就可以降低这种开销。然而，我们通常并不知道哪些操作符可能出问题，除非已经在实际运行中观测到错误、发现缺少装配信息，然后修改代码以启用装配跟踪，希望能再次观测到同样的错误。
在那种场景下，我们必须切换到调试模式并做好准备，以便更好地观测错误的第二次出现，这次捕获所有额外信息。
如果您能识别出应用中装配的、对可维护性至关重要的响应式链，就可以用 checkpoint() 操作符把两种技术结合起来。
您可以把这个操作符链式接入方法链中。checkpoint 操作符的工作方式类似于钩子版本，但只针对它在该特定链中的那一环。
还有一个 checkpoint(String) 变体，允许您为装配 traceback 添加一个唯一的 String 标识符。这样就省略了堆栈跟踪，而是依靠描述来识别装配点。checkpoint(String) 比常规的 checkpoint 施加的处理开销更小。
最后但同样重要的是，如果您想为 checkpoint 添加更通用的描述，但仍想依靠堆栈跟踪机制来识别装配点，可以使用 checkpoint("description", true) 版本来强制该行为。这样我们就回到了 traceback 的初始消息，并附加了一个 description，如下例所示：
Assembly trace from producer [reactor.core.publisher.ParallelSource], described as [descriptionCorrelation1234] : 1
reactor.core.publisher.ParallelFlux.checkpoint(ParallelFlux.java:215)
reactor.core.publisher.FluxOnAssemblyTest.parallelFluxCheckpointDescriptionAndForceStack(FluxOnAssemblyTest.java:225)
Error has been observed at the following site(s):
|_ ParallelFlux.checkpoint ⇢ reactor.core.publisher.FluxOnAssemblyTest.parallelFluxCheckpointDescriptionAndForceStack(FluxOnAssemblyTest.java:225)
descriptionCorrelation1234就是checkpoint中提供的描述。
描述可以是一个静态标识符、一段用户可读的描述，或一个更宽泛的关联 ID（例如，在 HTTP 请求场景下来自某个 header）。
当全局调试与 checkpoint 一起启用时，会采用全局调试的 traceback 风格，checkpoint 仅体现在「Error has been observed…」段落中。因此，在这种情况下，重量级 checkpoint 的名称是不可见的。
4. 生产环境可用的全局调试
Project Reactor 附带一个独立的 Java Agent，它会对您的代码进行插桩并添加调试信息，而无需为每次操作符调用捕获堆栈跟踪付出开销。其行为与激活调试模式（又称 traceback）非常相似，但没有运行时性能开销。
要在应用中使用它，您必须把它加为依赖。
下例展示如何在 Maven 中把 reactor-tools 加为依赖：
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-tools</artifactId>
// <1>
</dependency>
如果您使用 BOM，则无需指定
<version>。
下例展示如何在 Gradle 中把 reactor-tools 加为依赖：
dependencies {
compile 'io.projectreactor:reactor-tools'
}
它还需要显式初始化：
ReactorDebugAgent.init();
由于该实现会在类被加载时对其进行插桩，最佳放置位置是在 main(String[]) 方法中一切其他代码之前：
public static void main(String[] args) {
ReactorDebugAgent.init();
SpringApplication.run(Application.class, args);
}
如果无法急切地运行初始化，您也可以用 processExistingClasses() 重新处理已存在的类。例如，在 来自 TestExecutionListener 的 JUnit5 测试中，甚至在类的 static 初始化块中：
ReactorDebugAgent.init();
ReactorDebugAgent.processExistingClasses();
请注意，由于需要遍历所有已加载的类并应用转换，重新处理会花费几秒钟。只有在您发现某些调用点未被插桩时才使用它。
4.1. 局限性
ReactorDebugAgent 以 Java Agent 的形式实现，并使用 ByteBuddy 执行自附加（self-attach）。自附加在某些 JVM 上可能不起作用，详情请参阅 ByteBuddy 的文档。
4.2. 以 Java Agent 方式运行 ReactorDebugAgent
如果您的环境不支持 ByteBuddy 的自附加，可以把 reactor-tools 作为 Java Agent 运行：
java -javaagent:reactor-tools.jar -jar app.jar
4.3. 在构建时运行 ReactorDebugAgent
也可以在构建时运行 reactor-tools。为此，您需要把它作为 ByteBuddy 构建插桩的插件来应用。
该转换只会应用于您项目的类。classpath 中的库不会被插桩。
<dependencies>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-tools</artifactId>
// <1>
<classifier>original</classifier> // <2>
<scope>runtime</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>net.bytebuddy</groupId>
<artifactId>byte-buddy-maven-plugin</artifactId>
<configuration>
<transformations>
<transformation>
<plugin>reactor.tools.agent.ReactorDebugByteBuddyPlugin</plugin>
</transformation>
</transformations>
</configuration>
</plugin>
</plugins>
</build>
如果您使用 BOM，则无需指定
<version>。这里的
classifier很重要。
plugins {
id 'net.bytebuddy.byte-buddy-gradle-plugin' version '1.10.9'
}
configurations {
byteBuddyPlugin
}
dependencies {
byteBuddyPlugin(
group: 'io.projectreactor',
name: 'reactor-tools',
1
classifier: 'original', 2
)
}
byteBuddy {
transformation {
plugin = "reactor.tools.agent.ReactorDebugByteBuddyPlugin"
classPath = configurations.byteBuddyPlugin
}
}
如果您使用 BOM，则无需指定
version。这里的
classifier很重要。
5. 记录序列日志
除了堆栈跟踪调试与分析之外，您工具集中还有另一个强大的工具：追踪并记录异步序列中的事件。
log() 操作符正能做到这一点。把它链式接入序列后，它会窥视其上游 Flux 或 Mono 的每一个事件（包括 onNext、onError、onComplete，以及订阅、取消和请求）。
例如，假设我们已启用并配置了 Logback，并有一条类似 range(1,10).take(3) 的链。通过在 take 之前放置一个 log()，我们可以了解它是如何工作的、以及它向上游 range 传播了哪些事件，如下例所示：
Flux<Integer> flux = Flux.range(1, 10)
.log()
.take(3);
flux.subscribe();
这会打印出以下内容（通过 logger 的控制台 appender）：
10:45:20.200 [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription) 1
10:45:20.205 [main] INFO reactor.Flux.Range.1 - | request(3) 2
10:45:20.205 [main] INFO reactor.Flux.Range.1 - | onNext(1) 3
10:45:20.205 [main] INFO reactor.Flux.Range.1 - | onNext(2)
10:45:20.205 [main] INFO reactor.Flux.Range.1 - | onNext(3)
10:45:20.205 [main] INFO reactor.Flux.Range.1 - | cancel() 4
这里，除了 logger 自身的格式化器（时间、线程、级别、消息）之外，log() 操作符还会以自己的格式输出一些内容：
reactor.Flux.Range.1是日志的自动分类名，以防您在一条链中多次使用该操作符。它让您能区分记录的是哪个操作符的事件（这里是range）。您可以使用log(String)方法签名，用自己的自定义分类名覆盖该标识符。经过几个分隔字符后，实际事件被打印出来。这里我们看到一次onSubscribe调用、一次request调用、三次onNext调用和一次cancel调用。对于第一行onSubscribe，我们得到Subscriber的实现，它通常对应于该操作符特有的实现。方括号之间是额外信息，包括该操作符能否通过同步或异步融合被自动优化。在第二行，我们可以看到 take 把向上游的请求限制为 3。
然后 range 连续发出三个值。
在最后一行，我们看到
cancel()。
第二行（2）和最后一行（4）最值得关注。我们可以在那里看到 take 的实际作用。它利用背压，向源恰好请求预期数量的元素。在收到足够多的元素后，它通过调用 cancel() 告诉源不再需要更多元素。请注意，如果下游自身也使用了背压（例如只请求 1 个元素），take 操作符也会遵从（它在把请求从下游传播到上游时会将请求截顶）。