Spring 部落格

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

Reactor 除錯經驗

工程 | Sergei Egorov | 2019 年 3 月 28 日 | ...

在 Project Reactor 團隊中，我們相信您所依賴的函式庫的除錯體驗，與功能集或效能一樣重要。

今天，我們很高興宣布 Reactor 系列中的兩個新的實驗性專案！

BlockHound - 新手報到

最常見的新手錯誤之一是阻擋本應只執行非阻擋程式碼的 Java 執行緒（例如，Schedulers.parallel()）。
這是最有害的問題之一，因為您可能會阻擋不相關的處理，甚至建立死鎖！

考慮以下程式碼

Flux.range(0, Runtime.getRuntime().availableProcessors() * 2)
        .subscribeOn(Schedulers.parallel())
        .map(i -> {
            CountDownLatch latch = new CountDownLatch(1);

            Mono.delay(Duration.ofMillis(i * 100))
                .subscribe(it -> latch.countDown());

            try {
                latch.await();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            return i;
        })
        .blockLast();

執行此程式碼需要多長時間？ 1 秒？ 10 秒？
如果我告訴您它永遠不會退出並建立死鎖呢？原因如下

  1. 我們建立 N * 2 個訊號，其中 N 是我們的 JVM 可以使用的 CPU 數量。
  2. 我們使用 Schedulers.parallel 進行訂閱，這是一個限制為 N 個執行緒的 bounded pool。
  3. 對於每個訊號，我們為 parallel scheduler 排程另一個任務 (Mono.delay 隱式地使用 Schedulers.parallel)。
  4. 我們的邏輯假設延遲將很快被處理，並且 latch 最終將解除阻擋。
  5. 但是，所有 N 個執行緒都將等待它們的 latches，並且延遲任務將永遠不會被執行！

即使您沒有阻擋所有執行緒，而只阻擋了一些執行緒，您也會阻止其他不相關的任務前進。最可能的结果是效能會降低。

當您將舊的阻擋程式碼遷移到反應式方法時，問題尤其值得注意。即使是最有經驗的程式碼審查人員，當您的函數顏色相同時，也可能無法發現阻擋呼叫！

這就是我們創建 BlockHound 的原因，它是一個 Java Agent，用於檢測來自非阻擋執行緒的阻擋呼叫。與其他解決方案不同，它會檢測原始方法（甚至是 native 方法！），並且不允許以任何方式呼叫阻擋方法，即使使用反射也不行！

現在，如果我們按照文件中的描述將其添加到我們的應用程式中，我們將得到以下異常

java.lang.Error: Blocking call! sun.misc.Unsafe#park
  at reactor.BlockHound$Builder.lambda$new$0(BlockHound.java:154)
  at reactor.BlockHound$Builder.lambda$install$8(BlockHound.java:254)
  at reactor.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:43)
  at sun.misc.Unsafe.park(Unsafe.java)
  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
  at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
  at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
  at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
  at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
  at com.example.demo.BlockingCode.lambda$main$1(BlockingCode.java:24)

請注意，await 在內部呼叫 Unsafe#park 以實現等待邏輯。我們不希望我們的 Thread 被 parked 或阻擋，而 BlockHound 會保護我們免受這種情況的影響！

如果您想了解實作細節，請閱讀 how it works 頁面。
TL;DR：它包裝了原始方法，並且僅向它們添加了兩個方法呼叫。

您可以在測試或 QA/staging 環境中執行它，而不會損失效能。天啊，鑑於低開銷，您甚至可以嘗試在生產環境中執行它！ :)

BlockHound 適用於 Project Reactor 和 RxJava 2，您可以編寫您自己的整合。

Reactor Debug Agent - 生產環境就緒的組合回溯追蹤

由於反應式程式碼的功能性程式設計方面，除錯反應式程式碼有時可能具有挑戰性：您不是命令對資料執行什麼操作，而是聲明資料應如何流經系統。這意味著聲明和執行發生在不同的時間點。

您可以在 Simon 的精彩文章中閱讀更多相關資訊：https://spring.dev.org.tw/blog/2019/03/06/flight-of-the-flux-1-assembly-vs-subscription

在 Reactor 中，我們稱其為「組合時間」和「執行時間」。在組合時間，您透過呼叫 myFlux.map(i -> i * 2).filter(5 % i == 1).single() 和其他運算符來「設計」您的 pipeline。稍後，此「pipeline 定義」將用於處理 myFlux 發布的訊號。但是，當發生錯誤時會發生什麼？

你們中的一些人可能已經知道 Hooks.onOperatorDebug()。這是 reactor-core 中非常有用的 Hook。它將堆疊追蹤從

java.lang.IndexOutOfBoundsException: Source emitted more than one item
  at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
  at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129)
  at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107)
  at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94)
  at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186)
  at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131)
  at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114)
  at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68)
  at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58)
  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.util.concurrent.FutureTask.run(FutureTask.java:266)
  at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
  at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at java.lang.Thread.run(Thread.java:748)

轉換為

java.lang.IndexOutOfBoundsException: Source emitted more than one item
  at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
   ...
  at java.lang.Thread.run(Thread.java:748)
  Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoSingle] :
  reactor.core.publisher.Flux.single(Flux.java:7380)
  com.example.demo.Example.run(Example.java:13)
Error has been observed by the following operator(s):
  |_	Flux.single ⇢ com.example.demo.Example.run(Example.java:13)
  |_	Mono.subscribeOn ⇢ com.example.demo.Example.run(Example.java:14)

對於以下程式碼

9:  public class Example {
10:
11:     public static void run() {
12:        Flux.range(0, 5)
13:            .single() // <-- Aha!
14:            .subscribeOn(Schedulers.parallel())
15:            .block();
16:     }
17: }

如您所見，啟用除錯模式後，我們可以清楚地識別發生錯誤的組合操作。它就像堆疊追蹤，但（由於執行與組裝分離）是一個回溯追蹤。

您可能會想：「太棒了，現在我想在生產環境中使用它！」 - 我們也是。但是，當您使用 Hooks.onOperatorDebug() 時，我們必須在組合時進行重量級的堆疊追蹤，以在每次呼叫像 .map(...) 這樣的運算符時捕獲呼叫站點，即使您的程式碼永遠不會拋出錯誤！這是由於 Java 中缺少呼叫站點追蹤，唯一的替代方法是 new Exception().getStackTrace() 或 StackWalker（在 Java 9+ 中）。

顯然，我們不能在生產環境中使用這種方法，所以我們製作了一個工具！

來自 reactor-tools 專案的 ReactorDebugAgent 是一個 Java Agent，它可以幫助您除錯應用程式中的異常，而無需支付運行時成本（與 Hooks.onOperatorDebug() 不同）。

⚠️ 這個專案處於孵化階段，並且可能成為也可能不會成為獨立專案或 https://github.com/reactor/reactor-core 的模組。

它（透過位元組碼轉換）轉換鏈，例如

Flux.range(0, 5)
       .single()

到

Flux flux = Flux.range(0, 5);
flux = Hooks.addCallSiteInfo(flux, "Flux.range\n foo.Bar.baz(Bar.java:21)"));
flux = flux.single();
flux = Hooks.addCallSiteInfo(flux, "Flux.single\n foo.Bar.baz(Bar.java:22)"));

要啟用它，您需要先初始化 Agent

ReactorDebugAgent.init();

ℹ️ 由於實作會在載入類別時檢測您的類別，因此最好的放置位置是在您的 main(String[]) 方法中的所有其他內容之前

public static void main(String[] args) {
    ReactorDebugAgent.init();
    SpringApplication.run(Application.class, args);
}

結論

我們希望這些工具能讓您作為開發人員的生活更輕鬆，並讓您在使用 Project Reactor 時感到更舒適！

獲取 Spring 電子報

與 Spring 電子報保持聯繫

訂閱

領先一步

VMware 提供培訓和認證，以加速您的進度。

了解更多

取得支援

Tanzu Spring 在一個簡單的訂閱中提供 OpenJDK™、Spring 和 Apache Tomcat® 的支援和二進位檔案。

了解更多

即將舉辦的活動

查看 Spring 社群中所有即將舉辦的活動。

查看全部