Spring WebFlux アプリケーションをデバッグする方法

Spring WebFlux アプリケーションのデバッグは、特に複雑なリアクティブ ストリームを処理する場合、困難な作業になる可能性があります。スタック トレースが問題の根本原因を明確に示す従来のブロッキング アプリケーションとは異なり、リアクティブなアプリケーションはデバッグが困難になる可能性があります。コードのブロック、同時実行の問題、競合状態などの問題はすべて、診断が難しい微妙なバグを引き起こす可能性があります。

シナリオ

バグに対処するとき、それは必ずしもコード関連の問題ではありません。最近のリファクタリング、チームの変更、厳しい締め切りなど、一連の要因である可能性があります。実生活では、しばらく前に退職し、あなたが入社したばかりの人が作成した大規模なアプリケーションのトラブルシューティングを行うことはよくあることです。

ドメインとテクノロジーについて少し知っていても、人生が楽になるわけではありません。

以下のコード例では、最近チームに参加した人にとって、バグのあるコードがどのように見えるかを想像したいと思いました。

このコードのデバッグは、挑戦ではなく旅のように考えてください。根本的な原因は、リアクティブ アプリに慣れている人にとっては明らかです。ただし、以下のいくつかのプラクティスは、修正するのに非常に役立つ場合があります。

 @GetMapping("/greeting/{firstName}/{lastName}") public Mono<String> greeting(@PathVariable String firstName, @PathVariable String lastName) { return Flux.fromIterable(Arrays.asList(firstName, lastName)) .filter(this::wasWorkingNiceBeforeRefactoring) .transform(this::senselessTransformation) .collect(Collectors.joining()) .map(names -> "Hello, " + names); } private boolean wasWorkingNiceBeforeRefactoring(String aName) { // We don't want to greet with John, sorry return !aName.equals("John"); } private Flux<String> senselessTransformation(Flux<String> flux) { return flux .single() .flux() .subscribeOn(Schedulers.parallel()); }

つまり、このコードは次のようになります。パラメーターとして提供される名前の先頭に「Hello」を追加します。

あなたの同僚のジョンは、彼のラップトップですべてが動作すると言っています。それは本当だ：

 > curl localhost:8080/greeting/John/Doe > Hello, Doe

しかし、 curl localhost:8080/greeting/Mick/Jaggerのように実行すると、次のスタックトレースが表示されます。

 java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:134) ~[reactor-core-3.5.5.jar:3.5.5] Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: Error has been observed at the following site(s): *__checkpoint ⇢ Handler com.example.demo.controller.GreetingController#greeting(String, String) [DispatcherHandler] *__checkpoint ⇢ HTTP GET "/greeting/Mick/Jagger" [ExceptionHandlingWebHandler] Original Stack Trace: <18 internal lines> at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na] (4 internal lines)

いいですね、どちらのトレースも上記のコード サンプルにはつながりません。

明らかになるのは、1) GreetingController#greetingメソッドで発生し、2) クライアントが「HTTP GET "/greeting/Mick/Jagger を実行した」ということだけです。

.doOnError()

最初に試す最も簡単な方法は、グリーティング チェーンの最後に `.doOnError()` コールバックを追加することです。

 @GetMapping("/greeting/{firstName}/{lastName}") public Mono<String> greeting(@PathVariable String firstName, @PathVariable String lastName) { return Flux.fromIterable(Arrays.asList(firstName, lastName)) // <...> .doOnError(e -> logger.error("Error while greeting", e)); }

いい試みですが、ログには改善が見られません。

それにもかかわらず、Reactor の内部スタック トレース:

デバッグ中にdoOnError役立つ/役に立たないいくつかの方法を次に示します。

1. ロギング: doOnErrorを使用してエラー メッセージをログに記録し、リアクティブ ストリームで何が問題になったかについてより多くのコンテキストを提供できます。これは、多数のオペレーターを含む複雑なストリームで問題をデバッグする場合に特に役立ちます。

2. 回復: doOnError使用して、エラーから回復し、ストリームの処理を続行することもできます。たとえば、 onErrorResume使用して、エラーの場合にフォールバック値またはストリームを提供できます。

3. デバッグ: ほとんどの場合、 doOnErrorログで既に見たものを除いて、より良いスタック トレースを提供しません。これを優れたトラブルシューティング ツールとして信頼しないでください。

   
   

ログ（）

次のステップは、以前に追加したdoOnError() log()メソッド呼び出しに置き換えることです。それが得るのと同じくらい簡単です。 log()は、すべての Reactive Streams シグナルを監視し、デフォルトで INFO レベルのログにトレースします。

ログに表示される追加情報を見てみましょう。

どの Reactive メソッドが呼び出されたかを確認できます ( onSubscribe 、 request 、およびonError )。さらに、これらのメソッドがどのスレッド (プール) から呼び出されたかを知ることは、非常に役立つ情報です。ただし、それは私たちの場合には関係ありません。

log()メソッドを使用すると、ロギング ステートメントをストリームに追加できるため、データの流れの追跡と問題の診断が容易になります。 flatMap、サブチェーン、ブロッキング呼び出しなどでより複雑なデータ フローを使用している場合は、すべてをログに記録することで多くのメリットが得られます。毎日使うのにとても簡単で素敵なものです。しかし、根本的な原因はまだわかっていません。

Hooks.onOperatorDebug()

命令Hooks.onOperatorDebug()は、Reactor に、リアクティブ ストリーム内のすべてのオペレーターに対してデバッグ モードを有効にするよう指示し、より詳細なエラー メッセージとスタック トレースを可能にします。

公式ドキュメントによると：

この命令は、ランタイムごとに 1 回呼び出す必要があります。最適な場所の 1 つは、構成またはメイン クラスです。私たちのユースケースでは、次のようになります。

 public Mono<String> greeting(@PathVariable String firstName, @PathVariable String lastName) { Hooks.onOperatorDebug(); return // <...> }

Hooks.onOperatorDebug()を追加することで、最終的に調査を進めることができます。 Stacktrace の方がはるかに便利です。

42 行目には、 single()呼び出しがあります。

上にスクロールしないでください。senselessTransformation senselessTransformation次のようになります。

 private Flux<String> senselessTransformation(Flux<String> flux) { return flux .single() // line 42 .flux() .subscribeOn(Schedulers.parallel()); }

それが根本原因です。

single()は、Flux ソースから 1 つの項目を発行するか、複数の要素を持つソースに対してIndexOutOfBoundsExceptionを通知します。これは、メソッド内のフラックスが複数のアイテムを発行することを意味します。呼び出し階層を上に行くと、もともとFlux.fromIterable(Arrays.asList(firstName, lastName)) 2 つの要素を持つ Flux があることがわかります。

フィルタリング メソッドwasWorkingNiceBeforeRefactoring項目がJohnと等しい場合にフラックスから項目を削除します。これが、John という名前の大学でコードが機能する理由です。は。

Hooks.onOperatorDebug() 、ストリームがどのように処理されているかについてより詳細な情報を提供するため、複雑なリアクティブ ストリームをデバッグするときに特に役立ちます。ただし、デバッグ モードを有効にすると、アプリケーションのパフォーマンスに影響を与える可能性があるため (スタック トレースが読み込まれるため)、開発およびデバッグ中にのみ使用し、運用環境では使用しないでください。

チェックポイント

Hooks.onOperatorDebug()とほぼ同じ効果を最小限のパフォーマンスへの影響で達成するために、特別なcheckpoint()演算子があります。ストリームのそのセクションのデバッグ モードを有効にしますが、ストリームの残りの部分は影響を受けません。

フィルタリング後と変換後に 2 つのチェックポイントを追加しましょう。

 public Mono<String> greeting(@PathVariable String firstName, @PathVariable String lastName) { return Flux.fromIterable(Arrays.asList(firstName, lastName)) .filter(this::wasWorkingNiceBeforeRefactoring) /* new */ .checkpoint("After filtering") .transform(this::senselessTransformation) /* new */ .checkpoint("After transformation") .collect(Collectors.joining()) .map(names -> "Hello, " + names); }

ログを見てください：

このチェックポイントの内訳は、変換後として説明されている 2 番目のチェックポイントの後にエラーが観察されたことを示しています。実行中に最初のチェックポイントに到達していないという意味ではありません。だったのですが、エラーが出るようになったのは2回目以降です。そのため、 After filtering は表示されません。

DispatcherHandlerとExceptionHandlingWebHandlerから、内訳で言及されているさらに 2 つのチェックポイントも確認できます。呼び出し階層まで、設定した後に到達しました。

説明の他に、 checkpoint()メソッドの 2 番目のパラメーターとしてtrueを追加することで、Reactor にチェックポイントのスタック トレースを生成させることができます。生成されたスタック トレースによって、チェックポイントのある行が表示されることに注意してください。元の例外のスタックトレースは作成されません。説明を提供することでチェックポイントを簡単に見つけることができるため、あまり意味がありません。

結論

これらのベスト プラクティスに従うことで、デバッグ プロセスを簡素化し、Spring WebFlux アプリケーションの問題をすばやく特定して解決できます。経験豊富な開発者でも、リアクティブ プログラミングを始めたばかりでも、これらのヒントは、コードの品質と信頼性を向上させ、ユーザーにより良いエクスペリエンスを提供するのに役立ちます。