Spring 博客

所有帖子
工程
发布
新闻和事件

Flux 之飞翔 2 - 调试注意事项

工程 | Simon Baslé | 2019年4月16日 | ...

这篇博文是系列文章中的第二篇，旨在深入探讨Reactor更高级的概念和内部工作原理。

它源于我的 Flight of the Flux 演讲，我发现其内容更适合博文格式。

我将在其他文章发布时更新下表中的链接，但这是计划中的内容

  1. 组装与订阅
  2. 调试注意事项（本文）
  3. 跳跃线程和调度器
  4. 内部工作原理：工作窃取
  5. 内部工作原理：操作符融合

如果您错过了 Reactive Streams 和 Reactor 的基本概念介绍，请访问网站的学习部分和参考指南。

话不多说，让我们开始吧

在 Reactive 世界中调试

从命令式、阻塞式范式切换到反应式、非阻塞式范式会带来好处，但也会带来一些问题。其中之一就是调试体验。这是为什么呢？

主要是因为您已经习惯了依赖老式的 堆栈跟踪，但突然之间，由于反应式编程的异步特性，这个宝贵的工具变得不那么有价值了。但这并非反应式编程所特有：一旦引入异步代码，您就会在程序中创建一个边界，将调度代码和异步执行代码分隔开来。

用普通异步代码演示问题

我们以 ExecutorService 和 Future 为例（这里没有 Reactor 代码）

	private static void imperative() throws ExecutionException, InterruptedException {
		final ScheduledExecutorService executor =
				Executors.newSingleThreadScheduledExecutor();

		int seconds = LocalTime.now().getSecond();
		List<Integer> source;
		if (seconds % 2 == 0) {
			source = IntStream.range(1, 11).boxed().collect(Collectors.toList());
		}
		else if (seconds % 3 == 0) {
			source = IntStream.range(0, 4).boxed().collect(Collectors.toList());
		}
		else {
			source = Arrays.asList(1, 2, 3, 4);
		}

		executor.submit(() -> source.get(5))  //line 76
		        .get();
	}

这个例子有点做作，但假设我们代码中有三分之二的路径会导致异步任务抛出 IndexOutOfBoundsException... 堆栈跟踪会有多大帮助呢？

java.util.concurrent.ExecutionException: java.lang.ArrayIndexOutOfBoundsException: Index 5 out of bounds for length 4
	at java.base/java.util.concurrent.FutureTask.report(FutureTask.java:122)
	at java.base/java.util.concurrent.FutureTask.get(FutureTask.java:191)
	at Scratch.imperative(Scratch.java:77)
	at Scratch.main(Scratch.java:50)
Caused by: java.lang.ArrayIndexOutOfBoundsException: Index 5 out of bounds for length 4
	at java.base/java.util.Arrays$ArrayList.get(Arrays.java:4351)
	at Scratch.lambda$imperative$0(Scratch.java:76)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)

我们看到

  • Future 的 get() 方法抛出了 ExecutionException
  • 原因是 IndexOutOfBoundsException
  • 抛出异常的代码在 submit(() -> source.get(5)) lambda 的第 76 行
  • 它在 FutureTask 中执行，来自一个名为 ThreadPoolExecutor 的东西，它本身在 Thread 中运行...
  • 我们有两个潜在的源头可能导致这种情况，但不知道哪个是罪魁祸首（在调用 submit() 之前采用了哪条路径）。

用处不大 :-(

在 Reactor 中演示问题

如果我们寻找上面代码的 Reactor 等效项，我们可以得到以下代码

	private static void reactive() {
		int seconds = LocalTime.now().getSecond();
		Mono<Integer> source;
		if (seconds % 2 == 0) {
			source = Flux.range(1, 10)
			             .elementAt(5);
		}
		else if (seconds % 3 == 0) {
			source = Flux.range(0, 4)
			             .elementAt(5);
		}
		else {
			source = Flux.just(1, 2, 3, 4)
			             .elementAt(5);
		}

		source.subscribeOn(Schedulers.parallel())
		      .block(); //line 97
	}

这将触发以下堆栈跟踪

java.lang.IndexOutOfBoundsException
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.onComplete(MonoElementAt.java:153)
	at reactor.core.publisher.FluxArray$ArraySubscription.fastPath(FluxArray.java:176)
	at reactor.core.publisher.FluxArray$ArraySubscription.request(FluxArray.java:96)
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.request(MonoElementAt.java:92)
	at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186)
	at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131)
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.onSubscribe(MonoElementAt.java:107)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:53)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:59)
	at reactor.core.publisher.MonoElementAt.subscribe(MonoElementAt.java:59)
	at reactor.core.publisher.Mono.subscribe(Mono.java:3711)
	at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
	Suppressed: java.lang.Exception: #block terminated with an error
		at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:93)
		at reactor.core.publisher.Mono.block(Mono.java:1495)
		at Scratch.reactive(Scratch.java:97)
		at Scratch.main(Scratch.java:51)
  • 我们再次看到了 ArrayIndexOutOfBoundsException，这暗示源对于 MonoElementAt 运算符来说太短了
  • 我们看到它来自 onComplete，本身由 request 触发...以及 reactor.core.publisher 中的一堆其他步骤
  • 对这些 reactor 方法稍有熟悉，我们可能会推断出管道由 range (FluxRange.subscribe)、elementAt 和 subscribeOn 组成...
  • 抛出代码似乎是从 ThreadPoolExecutor 的工作 Thread 执行的
  • 线索在这里中断...

更糟糕的是，即使我们去掉了 subscribeOn，我们仍然无法发现是哪条可能的错误路径被触发了

	private static void reactiveNoSubscribeOn() {
		int seconds = LocalTime.now().getSecond();
		Mono<Integer> source;
		if (seconds % 2 == 0) {
			source = Flux.range(1, 10)
			             .elementAt(5);
		}
		else if (seconds % 3 == 0) {
			source = Flux.range(0, 4)
			             .elementAt(5);
		}
		else {
			source = Flux.just(1, 2, 3, 4)
			             .elementAt(5);
		}

		source.block(); //line 116
	}

给出堆栈跟踪

java.lang.IndexOutOfBoundsException
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.onComplete(MonoElementAt.java:153)
	at reactor.core.publisher.FluxArray$ArraySubscription.fastPath(FluxArray.java:176)
	at reactor.core.publisher.FluxArray$ArraySubscription.request(FluxArray.java:96)
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.request(MonoElementAt.java:92)
	at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49)
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.onSubscribe(MonoElementAt.java:107)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:53)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:59)
	at reactor.core.publisher.MonoElementAt.subscribe(MonoElementAt.java:59)
	at reactor.core.publisher.Mono.block(Mono.java:1494)
	at Scratch.reactiveNoSubscribeOn(Scratch.java:116)
	at Scratch.main(Scratch.java:52)
	Suppressed: java.lang.Exception: #block terminated with an error
		at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:93)
		at reactor.core.publisher.Mono.block(Mono.java:1495)
		... 2 more

这是因为，正如我们之前看到的，在组装和订阅之间存在一个额外的“边界”。线索只能追溯到订阅点（这里是 block()）:-(

因此，在异步世界中，使用堆栈跟踪进行分析和调试变得更加困难，在 Reactor 中甚至更困难（因为它既是异步的，又采用默认的懒惰组装与订阅方法）。但幸运的是，库中提供了工具来缓解这一事实。

让事情变得更好

回到经典：log

还记得你在命令式代码中散布 print 语句吗？它可能不如启动单步调试器酷炫，但有时它就是你需要的快速而粗暴的解决方案。

在 Reactor 中，您有 log() 操作符

  • 它会记录 Reactive Stream 信号：onNext、onComplete、onError（甚至包括 onSubscribe、cancel 和 request！）
  • 您可以对其进行调整，只允许这些信号中的一部分
  • 您也可以选择特定的 Logger

简而言之，log 是一个快速而粗糙的解决方案，可以轻松地鸟瞰序列中某个步骤的发生情况。在开发过程中尽情使用它，并可以为每个 log 调用指定一个“名称”以区分它们。

使用 log(String) 可以用来暗示哪个源导致了错误

	private static void log() {
		int seconds = LocalTime.now().getSecond();
		Mono<Integer> source;
		if (seconds % 2 == 0) {
			source = Flux.range(1, 10)
			             .elementAt(5)
			             .log("source A");
		}
		else if (seconds % 3 == 0) {
			source = Flux.range(0, 4)
			             .elementAt(5)
			             .log("source B");
		}
		else {
			source = Flux.just(1, 2, 3, 4)
			             .elementAt(5)
			             .log("source C");
		}

		source.block(); //line 138
	}

堆栈跟踪本身并没有更有趣（除了提到 MonoLogFuseable 类，但日志本身包含这个有趣的小细节）

17:01:23.711 [main] INFO  source C - | onSubscribe([Fuseable] MonoElementAt.ElementAtSubscriber)
17:01:23.716 [main] INFO  source C - | request(unbounded)
17:01:23.717 [main] ERROR source C - | onError(java.lang.IndexOutOfBoundsException)
17:01:23.721 [main] ERROR source C - 
java.lang.IndexOutOfBoundsException: null

至少我们得到了我们硬编码的 source C 标签...

使用调试模式丰富堆栈跟踪

Reactor 中可用的另一种方法是尝试在运行时堆栈跟踪中找回组装信息。

这可以通过 Hooks 类激活所谓的“调试模式”来完成

Hooks.onOperatorDebug();

它有什么作用？它使每个运算符实例化（即组装）都捕获一个堆栈跟踪并保留它以供以后使用。

如果 onError 到达某个运算符，它会将该组装堆栈跟踪附加到 onError 的 Throwable 上（作为抑制的 Exception）。因此，当您看到堆栈跟踪时，您将获得更完整的运行时和组装信息。

在调试模式开启的情况下，在我们之前的示例中，我们将能够看到采用了哪个组装路径以及实际处理了哪个源

	private static void hook() {
		Hooks.onOperatorDebug();
		try {
			int seconds = LocalTime.now().getSecond();
			Mono<Integer> source;
			if (seconds % 2 == 0) {
				source = Flux.range(1, 10)
				             .elementAt(5); //line 149
			}
			else if (seconds % 3 == 0) {
				source = Flux.range(0, 4)
				             .elementAt(5); //line 153
			}
			else {
				source = Flux.just(1, 2, 3, 4)
				             .elementAt(5); //line 157
			}

			source.block(); //line 160
		}
		finally {
			Hooks.resetOnOperatorDebug();
		}
	}

这会产生以下堆栈跟踪

java.lang.IndexOutOfBoundsException
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.onComplete(MonoElementAt.java:153)
(...)
	at reactor.core.publisher.Mono.block(Mono.java:1494)
	at Scratch.hook(Scratch.java:160)
	at Scratch.main(Scratch.java:54)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoElementAt] :
	reactor.core.publisher.Flux.elementAt(Flux.java:4367)
	Scratch.hook(Scratch.java:157)
Error has been observed by the following operator(s):
	|_	Flux.elementAt ⇢ Scratch.hook(Scratch.java:157)

注意最后一行了吗？太棒了 :-D

使用 checkpoint 降低成本

使用 Hooks.onOperatorDebug() 的一个缺点是，它会为应用程序中使用的每个运算符捕获组装堆栈跟踪。填充单个堆栈跟踪是一个昂贵的操作，因此不言而喻，这可能会对性能产生重大影响。因此，这仅建议在开发环境中使用。

幸运的是，如果您识别出代码库中容易出现那种源模糊不清的部分，您可以稍微降低成本。

通过使用 checkpoint() 操作符，可以仅在该代码库的特定点激活组装跟踪捕获。如果您使用 checkpoint(String) 为检查点提供一个唯一且有意义的名称，甚至可以完全不填充堆栈跟踪

	private static void checkpoint() {
		int seconds = LocalTime.now().getSecond();
		Mono<Integer> source;
		if (seconds % 2 == 0) {
			source = Flux.range(1, 10)
			             .elementAt(5)
			             .checkpoint("source range(1,10)");
		}
		else if (seconds % 3 == 0) {
			source = Flux.range(0, 4)
			             .elementAt(5)
			             .checkpoint("source range(0,4)");
		}
		else {
			source = Flux.just(1, 2, 3, 4)
			             .elementAt(5)
			             .checkpoint("source just(1,2,3,4)");
		}

		source.block(); //line 186
	}

这将产生以下堆栈跟踪

java.lang.IndexOutOfBoundsException
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.onComplete(MonoElementAt.java:153)
	at reactor.core.publisher.FluxArray$ArraySubscription.fastPath(FluxArray.java:176)
	at reactor.core.publisher.FluxArray$ArraySubscription.request(FluxArray.java:96)
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.request(MonoElementAt.java:92)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.request(FluxOnAssembly.java:438)
	at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49)
	at reactor.core.publisher.FluxOnAssembly$OnAssemblySubscriber.onSubscribe(FluxOnAssembly.java:422)
	at reactor.core.publisher.MonoElementAt$ElementAtSubscriber.onSubscribe(MonoElementAt.java:107)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:53)
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:59)
	at reactor.core.publisher.MonoElementAt.subscribe(MonoElementAt.java:59)
	at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:61)
	at reactor.core.publisher.Mono.block(Mono.java:1494)
	at Scratch.checkpoint(Scratch.java:186)
	at Scratch.main(Scratch.java:55)
	Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly site of producer [reactor.core.publisher.MonoElementAt] is identified by light checkpoint [source just(1,2,3,4)].

注意到最后一行了吗？is identified by light checkpoint [source just(1,2,3,4)].，这告诉我们罪魁祸首（因为我们为检查点使用了有意义的描述）。

结论

在本文中，我们了解到在异步编程中，堆栈跟踪可能不那么有用。这种影响因 Reactor 允许您以惰性方式构建反应式序列而进一步加剧。

我们研究了可能遇到的最坏情况以及减轻此问题的几种方法。

完整的代码可以在 这里 的 gist 中找到。

在下一篇中，我们将学习调度器以及如何在不同线程之间切换。

与此同时，祝您反应式编程愉快！

获取 Spring 新闻通讯

通过 Spring 新闻通讯保持联系

订阅

领先一步

VMware 提供培训和认证，助您加速进步。

了解更多

获得支持

Tanzu Spring 提供 OpenJDK™、Spring 和 Apache Tomcat® 的支持和二进制文件，只需一份简单的订阅。

了解更多

即将举行的活动

查看 Spring 社区所有即将举行的活动。

查看所有