問題タブ [project-reactor]

Spring 5.0.0 M1 から M2 にアップグレードした後、このコードは機能しなくなりました (クラスSseEventからServerSentEventクラスに変更しました)。

例外は次のとおりです。

次の問題があります: partitionId プロパティ (たとえば 0-10) を持つイベントのリストが与えられた場合、同じ partitionId を持つイベントが受信順に処理されるように、着信イベントを paritionId に従って分割したいと考えています。 . ほぼ均等に分散すると、(パーティションごとに) 10 個のイベントが並行して処理されることになります。

10 個のシングルスレッド ディスパッチャを作成し、適切なディスパッチャにイベントを送信する以外に、Project Reactor を使用して上記を達成する方法はありますか?

ありがとう。

サンプル プロジェクトは次の場所にあります: https://github.com/codependent/spring5-playground

JMS キューから受信したメッセージを、メッセージをイベント ストリームとして発行するリアクティブ コントローラーにブリッジしたいと考えています。

つまり、メッセージが到着し、サブスクライバーがいない場合、後でサブスクライバーが送信されたくないので、EmitterProcessor を使用しています。

これは私のMessageListener です:

最後に、Rest Controller:

その動作をテストするために、このコントローラー メソッドを追加して、キューへの挿入をシミュレートしました。

ロードしたアプリケーションを起動した直後http://localhost:8080/accounts/1/alerts/live2、ブラウザはデータを待ちます。

次に、いくつかのメッセージを公開しますhttp://localhost:8080/mock/accounts/1/alerts/put。

しかし、どれもブラウザに到達しません。これは最終的に 500 エラー (ログなし) で終了します。

手動で再試行した後、データの受信を開始します...

...しかし、他の多くの場合、何も得られません。

まず第一に、私は非同期処理の分野にかなり慣れていません。私の現在のプロジェクトでは、Spring Boot をプロジェクト リアクター (具体的には Eventbus) と共に使用して、非同期処理を実行しています。イベントバスを使用すると、システムのスケーラビリティも向上すると思います。

これまで、EventBus の使用は、何かを返さない EventBus コンシューマーで何らかの処理を行う場合にかなり制限されていました。構成とプロセッサの例は次のとおりです。

//設定ファイル

//消費者

これまでは、accept メソッドの戻り値の型が void で問題ありませんでした。しかし、現在、プロセッサメソッドから応答を返したいシナリオがあります。または、処理中にエラーが発生した場合は適切な例外をスローする必要があり、いずれの場合も、応答/例外をポイントに戻す必要があります。呼び出し。

これはリアクターを使用して行うことができますか? はいの場合、これの簡単な例を教えてください。Promise について読んだことがありますが、私のケースに似た例が見つかりません。

Reactor-core と Spring 5 は、Mono/Flux パブリッシャー用の新しい非同期 WebClient を導入しました。

非同期性を提供するためにRestTemplateリクエストをObservableとsubscribeOnでラップするのは非常に一般的であるため、RxJava用の非同期クライアントがあるかどうか疑問に思っていました。

https://github.com/ReactiveX/RxApacheHttpを見つけましたが、サポートされていないようです。最後のコミットは 2014 年のもので、テストでも機能していません。

RxJava または Reactive Stream でコンテンツベースのルーティングを行う慣用的な方法を知っている人はいますか?

map filter や transform などの標準演算子を使用して、リアクティブ メッセージ処理パイプラインを作成する方法を確認できます。

しかし、演算子を使用して複数の可能なパイプラインを切り替える理由が 1 つわかりません。
入力マップまたはオブジェクトの内容に応じて異なるパイプラインが選択されるように、1 つの初期入力オブジェクトまたはマップに基づいて、多くの処理パイプラインから 1 つをトリガーできるようにしたいと考えています。

アイデアは、外部 API に対していくつかの可能なリクエストを作成する必要があり、トリガーされる API とパイプラインを決定するものは、コントローラーへの入力オブジェクトに依存するということです。

これは Spring 統合または Camel で実行できることはわかっていますが、私のユースケースでは、ネットワーク上のレイテンシーを処理できるので、統合エンジンの完全な柔軟性は必要ありません。

よろしくお願いします

APIは、処理する必要があるデータのようなオブジェクトの配列を持つ JSON オブジェクトで応答します。{"items":[{...},{...},...],...問題は、配列に + 1.000 個のオブジェクトがあり、エンドポイントの応答が遅いため、ストリーミングが完了するまでに数秒かかることです。また、Apache HttpClientを使用してエンドポイントを消費し、Jacksonを使用して応答をシリアル化する「ブロッキング」プログラミング モデルを使用しています。 、配列を含むすべてのオブジェクトを解析するために完全なボディを取得するのを待ちます。要素が処理されたら、使用されているメモリを解放するために GC に任せることができるため、メモリ使用量を節約できます。要素によって。

データが来ている間にReactor プロジェクトで応答を使用するにはどうすればよいですか? 使用している HTTP クライアントやシリアライザーを変更しても問題ありません。また、Spring 4.3 の Spring MVC で Java 8 を使用していますが、プロジェクトを Spring の今後のリリースに移行できます。

注: エンドポイントはサード パーティの API であり、HTTP 1.1 サーバーへの単純な GET です。ページネーションの追加など、その動作方法を変更することはできません。

Tomcat サーバーが組み込まれた Spring を使用して、Web ソケット サブスクリプションを適切に機能させるのに問題があります。

このストンプ チャット アプリケーションを実行すると、次のエラーが表示されます。

次の依存関係を追加しようとしました：

しかし、それは同じエラーを出しました。どんな助けでも大歓迎です。