問題タブ [backpressure]

を使用して別のスレッドで Java から Swing イベント キューをフラッディングするとSwingUtilities.invokeLater(..)、イベント ディスパッチ スレッドが追いつかなくなり、EventQueue がいっぱいになり、アプリケーションの動作が遅くなります。

たとえば...次の例では、画面に表示されている整数の数しか表示されていません（mac os-x yosemite、java 8 update 25）。

GUIのすべての変更を実際に見ることができるように、「物事を遅くする」方法はありますか? ラベルrepaintを呼び出した後、どこかで何らかのメソッドを呼び出すようなものですか?setText(..)

私の目はピクセルの変化よりも「遅い」と予想されるため、激しいちらつきはまったく問題ありません。現時点では、この例は「奇妙な」動作をしており、最後に 1 秒間 1070201 のような特定の数値が表示され、次に 1 秒間 1531452 が表示されます。

もちろん、イベント キューが「いっぱい」になるのを防ぐために、SwingUtilities.invokeAndWait(..)代わりにSwingUtilities.invokeLater(..)「生成」メイン スレッドへのバックプレッシャーを使用して確立することもできます。他の方法はありませんか？

次のコードは、ファイルを 1 行ずつ読み取り、各行のタスクを作成してから、executor のキューに入れます。エグゼキューターのキューがいっぱいになると、ファイルからの読み取りは、再びスペースができるまで停止します。

SOでいくつかの提案を見ましたが、それらはすべて、ファイルのコンテンツ全体をメモリに読み込むか、次善のスケジューリングを必要とします(たとえば、100行を読み取り、それらを並行して処理し、それが終了した後にのみ、次の100行を読み取ります) . また、これには Akka のようなライブラリも使用したくありません。

これらの欠点なしでこれを達成するScalaの方法は何ですか?

の例示的な定義executorWithBoundedQueue

にrx.Observableタスクの進行状況を出力する がありますonNext()。onNext()放出が非常に速く発生してObserverが追いつかず、背圧が発生することがあります。からの最新の放出のみをバッファリングすることで、背圧を処理したいと思いObservableます。

例えば：

• Observable1を発行し、 1をObserver受け取ります。
• 1をObserverまだ処理している間に、 2、3、および4を発行します。Observable
• Observer処理1を終了し、処理4を開始します(放出2と3はドロップされます)。

これは、Rx Observable で進行状況を処理する一般的なケースのようです。これは、通常、UI を最新の進行状況情報で更新することだけに関心があるためです。しかし、私はこれを行う方法を理解することができませんでした。

RxJavaでこれを達成する方法を知っている人はいますか?

私が達成しようとしているのは、akka ストリームを使用した同期フィードバック ループのようなものを実装することです。

を持っているとしましょうFlow[Int].filter(_ % 5 == 0)。のストリームIntをこのフローにブロードキャストし、タプルをその直後に圧縮すると、次のような結果が得られます

Option[Int]次の要素をプッシュした後にフローが要素を放出したかどうかを示すを放出する方法はありますか?

DetachedStageフローが前Flowの段階で引っ張られるたびに、私は彼が次の要素を必要とすることを知っていました。背後のステージが要素を受け取らなかった場合、それは None でした。

残念ながら、結果は良くなく、多くの順位がずれています。

補足事項

フィルタ フローは単なる例です。これは非常に長いフローになる可能性があり、そのフローのOptionすべての段階で を発行する機能を提供することはできません。そのため、フローが次をプッシュしたかどうかを実際に知る必要があります。代わりにダウンストリームから次をリクエストしました

と も試してみましたがconflate、expandこれらは結果の位置オフセットでさらに悪化しています

構成で変更したことの 1 つは、フローのinitialandmaxバッファーでした。これにより、示された要求が実際にプッシュした要素の後にあることを確認できます。

この問題を解決する方法についていくつかの提案をいただければ幸いです。

rx Java でバックプレッシャーを使用して、Android アプリケーションで無限スクロールを作成しようとしています。外部サービスを要求された回数だけ呼び出すようにしたい（呼び出した後request(1)）。しかし、flatmap を使用した後は、すべてsubscribe16 ページが読み込まれます。

私のコードの下に期待される結果があります。最初のリクエストが原因でほぼすべてのテストが失敗します (n=16 の場合)

内部ノード間通信に現在 HTTP/REST を使用しているマイクロサービス アーキテクチャの最適化を検討しています。

1 つのオプションは、バックプレッシャ機能をサービスに実装することです (たとえば、Quasar のようなものをスタックに統合することによって)。これは間違いなく物事を改善するでしょう。しかし、いくつかの課題があります。1 つは、非同期クライアント スレッドが一時的 (メモリ内) であり、クライアントに障害 (クラッシュ) が発生すると、これらの再試行スレッドが失われることです。2 つ目は、理論的には、ターゲット サーバーがしばらくダウンしている場合、クエーサー ファイバーであってもスレッドが最終的に制限されるため、クライアントは再試行を試みて最終的に OOM に到達する可能性があります。

少し偏執的であることはわかっていますが、キューベースの代替手段が非常に大規模な場合により有利になるかどうか疑問に思っています。

ただし、a) キューが中央で管理され、クライアント JVM から切り離されていること、b) キューが永続的であるため、クライアントやターゲット サーバーがダウンした場合、処理中のメッセージがないことを除きます。失われます。

もちろん、キューの欠点は、ホップが多くなり、システムの速度が低下することです。しかし、Quasar の ROI がピークに達し、一元化された耐久性のあるキューがスケーリングと HA にとってより重要になるスイート スポットがおそらくあると思います。

私の質問は：

このトレードオフは議論されましたか? サービス内通信に集中型の外部キュー/ルーター アプローチを使用することに関する論文はありますか。

TL;DR; この質問はおそらく次のように表現できることに気付きました。

「マイクロサービス アーキテクチャ内で直接 HTTP を使用するのではなく、メッセージ バス ベースのサービス内通信を使用するのが適切なのはいつですか。」

特にこのwikiを読んで、リアクティブバックプレッシャー処理の問題に慣れようとしています: https ://github.com/ReactiveX/RxJava/wiki/Backpressure

バッファ パラグラフには、より複雑なサンプル コードがあります。

私の理解が正しければ、バッファー オペレーター用にデバウンスされた信号ストリームを生成することで、バースト ソース ストリームを効果的にデバウンスしています。

しかし、なぜここで publish 演算子と refcount 演算子を使用する必要があるのでしょうか? それらを単にドロップすると、どのような問題が発生しますか? RxJava Observables はデフォルトでマルチキャストに対応していませんか?