调试Reactive Streams可能是我们开始使用这些数据结构后必须面对的主要挑战之一。
考虑到Reactive Streams在过去几年中越来越受欢迎，了解我们如何有效地执行此任务是个好主意。
让我们首先使用Reactive Streams设置项目，看看为什么这通常很麻烦。
我们想要模拟一个实际情况，其中运行了几个异步进程，并且我们在代码中引入了一些最终会触发异常的缺陷。
为了理解全局，我们将提到我们的应用程序将使用和处理简单Foo对象流，这些对象只包含id、formattedName和quantity字段。
现在，让我们检查一个片段以及当出现未处理的错误时它生成的输出：
public void processFoo(Flux<Foo> flux) {
flux = FooNameHelper.concatFooName(flux);
flux = FooNameHelper.substringFooName(flux);
flux = FooReporter.reportResult(flux);
flux.subscribe();
}
public void processFooInAnotherScenario(Flux<Foo> flux) {
flux = FooNameHelper.substringFooName(flux);
flux = FooQuantityHelper.divideFooQuantity(flux);
flux.subscribe();
}
运行我们的应用程序几秒钟后，我们会看到它会不时记录异常。
仔细查看其中一个错误，我们会发现类似于此的内容：
Caused by: java.lang.StringIndexOutOfBoundsException: String index out of range: 15
at j.l.String.substring(String.java:1963)
at com.baeldung.debugging.consumer.service.FooNameHelper
.lambda$1(FooNameHelper.java:38)
at r.c.p.FluxMap$MapSubscriber.onNext(FluxMap.java:100)
at r.c.p.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
at r.c.p.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:275)
at r.c.p.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:849)
at r.c.p.Operators$MonoSubscriber.complete(Operators.java:1476)
at r.c.p.MonoDelayUntil$DelayUntilCoordinator.signal(MonoDelayUntil.java:211)
at r.c.p.MonoDelayUntil$DelayUntilTrigger.onComplete(MonoDelayUntil.java:290)
at r.c.p.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:118)
at r.c.s.SchedulerTask.call(SchedulerTask.java:50)
at r.c.s.SchedulerTask.call(SchedulerTask.java:27)
at j.u.c.FutureTask.run(FutureTask.java:266)
at j.u.c.ScheduledThreadPoolExecutor$ScheduledFutureTask
.access$201(ScheduledThreadPoolExecutor.java:180)
at j.u.c.ScheduledThreadPoolExecutor$ScheduledFutureTask
.run(ScheduledThreadPoolExecutor.java:293)
at j.u.c.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at j.u.c.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at j.l.Thread.run(Thread.java:748)
基于根本原因，并注意到堆栈跟踪中提到的FooNameHelper类，我们可以想象在某些情况下，我们的Foo对象正在使用比预期更短的formattedName 值进行处理。
当然，这只是一个简化的案例，解决方案似乎相当明显。
但是让我们假设这是一个真实案例场景，如果没有一些上下文信息，异常本身并不能帮助我们解决问题。
异常是作为processFoo或processFooInAnotherScenario方法的一部分触发的吗？
在到达此阶段之前，其他前面的步骤是否影响了formattedName字段？
日志条目无法帮助我们找出这些问题。
更糟糕的是，有时甚至不会从我们的功能中抛出异常。
例如，假设我们依赖反应式存储库来保存我们的Foo对象。如果此时错误上升，我们甚至可能不知道从哪里开始调试代码。
我们需要工具来有效地调试反应流。
确定我们的应用程序正在发生什么的一个选项是使用我们喜欢的IDE启动调试会话。
我们必须设置几个条件断点，并在流中的每个步骤执行时分析数据流。
实际上，当我们有大量的被动进程在运行和共享资源时,这可能是一项繁琐的任务。
此外，在许多情况下，出于安全原因，我们无法启动调试会话。
有时，我们可以通过提供Consumer作为subscribe方法的第二个参数来添加有用的上下文信息：
public void processFoo(Flux<Foo> flux) {
// ...
flux.subscribe(foo -> {
logger.debug("Finished processing Foo with Id {}", foo.getId());
}, error -> {
logger.error(
"The following error happened on processFoo method!",
error);
});
}
注意：值得一提的是，如果我们不需要对subscribe方法进行进一步处理，我们可以在发布者上链接doOnError函数：
flux.doOnError(error -> {
logger.error("The following error happened on processFoo method!", error);
}).subscribe();
现在我们将对错误的来源提供一些指导，即使我们仍然没有太多关于生成异常的实际元素的信息。
Reactor库提供了一个hook类，它允许我们配置Flux和Mono操作符的行为。
通过添加以下语句，我们的应用程序将检测对发布者方法的调用，包装运算符的构造，并捕获堆栈跟踪：
Hooks.onOperatorDebug();
这样就可以默认启用Thymeleaf - 无需额外配置。
调试模式激活后，我们的异常日志将包含一些有用的信息：
16:06:35.334 [parallel-1] ERROR c.b.d.consumer.service.FooService
- The following error happened on processFoo method!
java.lang.StringIndexOutOfBoundsException: String index out of range: 15
at j.l.String.substring(String.java:1963)
at c.d.b.c.s.FooNameHelper.lambda$1(FooNameHelper.java:38)
...
at j.l.Thread.run(Thread.java:748)
Suppressed: r.c.p.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [reactor.core.publisher.FluxMapFuseable] :
reactor.core.publisher.Flux.map(Flux.java:5653)
c.d.b.c.s.FooNameHelper.substringFooName(FooNameHelper.java:32)
c.d.b.c.s.FooService.processFoo(FooService.java:24)
c.d.b.c.c.ChronJobs.consumeInfiniteFlux(ChronJobs.java:46)
o.s.s.s.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:84)
o.s.s.s.DelegatingErrorHandlingRunnable
.run(DelegatingErrorHandlingRunnable.java:54)
o.u.c.Executors$RunnableAdapter.call(Executors.java:511)
o.u.c.FutureTask.runAndReset(FutureTask.java:308)
Error has been observed by the following operator(s):
|_ Flux.map ⇢ c.d.b.c.s.FooNameHelper
.substringFooName(FooNameHelper.java:32)
|_ Flux.map ⇢ c.d.b.c.s.FooReporter.reportResult(FooReporter.java:15)
我们可以看到，第一部分保持相对相同，但以下部分提供了有关以下内容的信息：
- 发布者的程序集跟踪 - 这里我们可以确认错误最初是在processFoo方法中生成的。
- 在第一次触发错误之后观察到错误的运算符，以及链接它们的用户类。
注意：在这个例子中，主要是为了清楚地看到这一点，我们在不同的类上添加操作。
我们可以随时打开或关闭调试模式，但它不会影响已经实例化的Flux和Mono对象。
要记住的另一个方面是即使在流上运行不同的线程，也会正确生成程序集跟踪。
我们来看看下面的例子：
public void processFoo(Flux<Foo> flux) {
flux = flux.publishOn(Schedulers.newSingle("foo-thread"));
// ...
flux = flux.publishOn(Schedulers.newSingle("bar-thread"));
flux = FooReporter.reportResult(flux);
flux.subscribeOn(Schedulers.newSingle("starter-thread"))
.subscribe();
}
现在，如果我们检查日志，我们会理解在这种情况下，第一部分可能会稍微改变，但最后两部分保持相同。
第一部分是线程堆栈跟踪，因此它只显示特定线程执行的操作。
正如我们所看到的，当我们调试应用程序时，这不是最重要的部分，因此这种更改是可以接受的。
在每个单一的反应过程中检测和生成堆栈跟踪都是昂贵的。
因此，我们应该只在关键情况下实施前一种方法。
无论如何，Reactor提供了一种在单个关键进程上启用调试模式的方法，这样可以减少内存消耗。
我们指的是检查点操作员：
public void processFoo(Flux<Foo> flux) {
// ...
flux = flux.checkpoint("Observed error on processFoo", true);
flux.subscribe();
}
请注意，以这种方式，将在检查点阶段记录程序集跟踪：
Caused by: java.lang.StringIndexOutOfBoundsException: String index out of range: 15
...
Assembly trace from producer [reactor.core.publisher.FluxMap],
described as [Observed error on processFoo] :
r.c.p.Flux.checkpoint(Flux.java:3096)
c.b.d.c.s.FooService.processFoo(FooService.java:26)
c.b.d.c.c.ChronJobs.consumeInfiniteFlux(ChronJobs.java:46)
o.s.s.s.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:84)
o.s.s.s.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
j.u.c.Executors$RunnableAdapter.call(Executors.java:511)
j.u.c.FutureTask.runAndReset(FutureTask.java:308)
Error has been observed by the following operator(s):
|_ Flux.checkpoint ⇢ c.b.d.c.s.FooService.processFoo(FooService.java:26)
我们应该在反应链的末尾实施检查点方法。
否则，操作员将无法观察下游发生的错误。
另外，请注意，库提供了重载方法。我们可以避免：
- 如果我们使用no-args选项，则指定观察到的错误的描述
- 通过仅提供自定义描述来生成填充堆栈跟踪（这是最昂贵的操作）
最后，Reactor发布商提供了一种在某些情况下可能会派上用场的方法。
通过在我们的反应链中调用log方法，应用程序将使用它在该阶段具有的状态记录流中的每个元素。
让我们在我们的例子中尝试一下：
public void processFoo(Flux<Foo> flux) {
flux = FooNameHelper.concatFooName(flux);
flux = FooNameHelper.substringFooName(flux);
flux = flux.log();
flux = FooReporter.reportResult(flux);
flux = flux.doOnError(error -> {
logger.error("The following error happened on processFoo method!", error);
});
flux.subscribe();
}
并检查日志：
INFO reactor.Flux.Map.1 - onSubscribe(FluxMap.MapSubscriber)
INFO reactor.Flux.Map.1 - request(unbounded)
INFO reactor.Flux.Map.1 - onNext(Foo(id=0, formattedName=theFo, quantity=8))
INFO reactor.Flux.Map.1 - onNext(Foo(id=1, formattedName=theFo, quantity=3))
INFO reactor.Flux.Map.1 - onNext(Foo(id=2, formattedName=theFo, quantity=5))
INFO reactor.Flux.Map.1 - onNext(Foo(id=3, formattedName=theFo, quantity=6))
INFO reactor.Flux.Map.1 - onNext(Foo(id=4, formattedName=theFo, quantity=6))
INFO reactor.Flux.Map.1 - cancel()
ERROR c.b.d.consumer.service.FooService
- The following error happened on processFoo method!
...
我们可以在此阶段轻松查看每个Foo对象的状态，以及在异常发生时框架如何取消流。
当然，这种方法也很昂贵，我们必须适度使用它。
如果我们不知道正确调试应用程序的工具和机制，我们可能会花费大量时间和精力来解决问题。
如果我们不习惯处理被动和异步数据结构，那么尤其如此，我们需要额外的帮助来弄清楚事情是如何工作的。
