内容简介:案例概述调试考虑到Reactive Streams在过去几年中越来越受欢迎,了解我们如何有效地执行此任务是个好主意。
案例概述
调试 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对象的状态,以及在异常发生时框架如何取消流。
当然,这种方法也很昂贵,我们必须适度使用它。
案例结论
如果我们不知道正确调试应用程序的工具和机制,我们可能会花费大量时间和精力来解决问题。
如果我们不习惯处理被动和异步数据结构,那么尤其如此,我们需要额外的帮助来弄清楚事情是如何工作的。
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网
猜你喜欢:- iOS常用调试方法:断点调试
- 断点调试和日志调试之间的平衡点:函数计算调试之 Python 篇
- .NET高级调试系列-Windbg调试入门篇
- VisualStudio 通过外部调试方法快速调试库代码
- GDB 调试 Mysql 实战(二)GDB 调试打印
- 使用gdb调试工具上手调试php和swoole源码
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。