Spring 部落格

所有文章
工程
發布
新聞與活動

Flux 2 的飛行 - 偵錯注意事項

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

這篇部落格文章是系列文章中的第二篇，旨在深入探討 Reactor 更進階的概念和內部運作方式。

它源自我的 Flux 的飛行 演講，我發現其內容更適合部落格文章的形式。

當其他文章發布時，我會更新下表中的連結，但以下是計劃的內容

  1. 組裝與訂閱
  2. 偵錯注意事項（本文）
  3. 跳躍執行緒與排程器
  4. 內部運作方式：工作竊取
  5. 內部運作方式：運算子融合

如果您錯過了 Reactive Streams 的介紹和 Reactor 的基本概念，請前往網站的 學習區 和 參考指南。

廢話不多說，讓我們開始吧

反應式世界中的偵錯

從命令式、阻塞式範例轉換到反應式、非阻塞式範例帶來了諸多好處，但也伴隨著一些注意事項。其中之一就是偵錯體驗。為什麼會這樣呢？

主要是因為您已經習慣依賴好用的 堆疊追蹤，但突然之間，由於反應式程式設計的非同步特性，這個寶貴的工具變得不那麼有價值了。這並非反應式程式設計所獨有：只要您引入非同步程式碼，您就會在程式中建立一個邊界，分隔 排程 程式碼和 非同步執行 程式碼。

使用原生非同步程式碼示範問題

讓我們以 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() 運算子

  • 它記錄反應式串流訊號：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 社群中所有即將到來的活動。

查看全部