在Spring 5中调试Reactive Streams
flx 2019-06-30
调试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启动调试会话。
我们必须设置几个条件断点，并在流中的每个步骤执行时分析数据流。
实际上，当我们有大量的被动进程在运行和共享资源时,这可能是一项繁琐的任务。
此外，在许多情况下，出于安全原因，我们无法启动调试会话。
使用doOnError方法或使用订阅参数记录信息有时，我们可以通过提供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的全局调试配置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对象的状态，以及在异常发生时框架如何取消流。
当然，这种方法也很昂贵，我们必须适度使用它。
案例结论如果我们不知道正确调试应用程序的工具和机制，我们可能会花费大量时间和精力来解决问题。
如果我们不习惯处理被动和异步数据结构，那么尤其如此，我们需要额外的帮助来弄清楚事情是如何工作的。
相关推荐
LOGGER.info("compose executed");return stringFlux.map(e -> e + "$");LOGGER.info("flatMap executed&qu