問題タブ [lmax]

LMAX と、RingBuffer と呼ばれるこの素晴らしい概念について紹介しました。そのため、1 つのスレッドのみでリングバッファーに書き込む場合、複数のプロデューサーを使用するよりもパフォーマンスが優れていると人々は言います...

ただし、典型的なアプリケーションがリングバッファーへの書き込みに1つのスレッドのみを使用する可能性があるとは本当に思いません... lmaxがどのようにそれを行っているかを本当に理解していません(もしそうなら)。たとえば、N 人の異なるトレーダーが取引所に注文を出すとします。これらはすべて、注文に変換されてリングバッファーに入れられる非同期の要求です。1 つのスレッドを使用してそれらをどのように書き込むことができるでしょうか?

質問 1. 何かが欠けているか、誤解している可能性がありますが、N 個の同時プロデューサーが存在する場合、それらを 1 つにマージし、互いにロックしないようにするにはどうすればよいですか?

質問 2. rxJava オブザーバブルを思い出しました。N 個のオブザーバブルを取得し、Observable.mergeを使用してそれらを 1 つにマージできましたが、何らかの方法でロックをブロックまたは維持しているのだろうか?

LMAX Disruptor Framework から WorkerPool の例を探しています。アイデアは、イベントが一度だけ消費されることを確認したいときに使用することです (サブスクライブ可能なチャネルではなく、ポーリング可能なチャネル)

LMAX ディストラプターを使用して、プログラムの 1 つによって生成されたコンテンツをバッファリングし、それらをレコードのバッチとして別のプログラムに公開しようとしました (まあ、まだ消費者のバッチ処理の部分を完了できません)。ただし、レコードのバッチ処理を使用しなくても、正常に機能します。しかし、私の問題は、私が使用したにもかかわらず

例の 1 つに示されているように、プログラムの実行は停止しません。これらのメソッドの下のステートメントも実行します。印刷するとき

true を返します。誰かがこれで私を助けることができます...

編集

私は最近 LMAX Disruptor について学び、いくつかの実験を行っています。私を困惑させていることの1つは、のハンドラーメソッドのendOfBatchパラメーターです。次のコードを検討してください。まず、私が呼び出すダミーのメッセージとコンシューマー クラスと:onEventEventHandlerTest1Test1Worker

実際の作業の代わりに 500 ミリ秒の遅延を設定したことに注意してください。また、コンソールにシーケンス番号を出力しています

そして、（プロデューサーとして機能している）ドライバークラスが呼び出されましたDisruptorTest：

ここでは、必要なものを初期化した後、10 個のメッセージを (バッファ サイズ 8) にフィードしRingBuffer、いくつかのことを監視しようとしています - の次のスロットを要求するためのプロデューサーの遅延とRingBuffer、コンシューマー側、および特定のシーケンスがバッチの最後と見なされているかどうか。

さて、興味深いことに、各メッセージの処理に 500 ミリ秒の遅延が関係しているため、これが出力として得られるものです。

ただし、500 ミリ秒の待機時間を削除すると、次のようになります。

したがって、特定のメッセージがバッチの最後にあると見なされるかどうか (つまり、バッチのサイズ) は、コンシューマーのメッセージ処理の遅延に影響されているようです。ここで私はばかげているかもしれませんが、それはどうあるべきですか？その背後にある理由は何ですか？とにかく、一般的にバッチサイズを決定するものは何ですか? 前もって感謝します。私の質問で不明な点があれば教えてください。

私はワーカープールの例で作業していて、さまざまな WaitStrategies を試していました。TimeoutBlockingWaitStrategy を試すと、エラーが発生します。これがプログラムとコールスタックです。

エラーは次のとおりです

私のコードにバグがあるかどうか疑問に思っています。どんな助けでも本当に感謝しています。また、誰かがディスラプターで TimeoutBlockingWaitStrategy の目的を説明できれば、それは素晴らしいことです。

以下の (部分的な) スタック トレースを理解しようとしています。春のプロジェクトリアクター2.0.4を使用。

タスクを作成し、ワーカー スレッドにタスクを割り当てるタスク スケジューラ スレッドを取得しました。この時点で、アプリケーションはハングします (ワーカー スレッドは何もログに記録しない、別の時間に取得されたスレッドダンプは常に同じなど)。

次の行に基づいて、リング バッファがいっぱいであると言えますか。at reactor.jarjar.com.lmax.disruptor.MultiProducerSequencer.next(MultiProducerSequencer.java:136)

リング バッファがいっぱいの場合、ワーカー スレッドがこれらのタスクのいずれも取得しないのはなぜですか? リングバッファの状態が壊れることはありますか?

リング バッファーとイベント ハンドラーがあります。これは、メッセージング サーバーの受信バッファーとして使用されます。

私の偶数ハンドラー内では、いくつかの AMQP サービス呼び出しが行われ、その呼び出しがタイムアウトなしで待機することがある (AMQP サービスを呼び出すアンダーライン プロトコル ライブラリのバグ)。そして 2000tps では、これにより 4096 サイズのリング バッファが一瞬で FULL になりました。前述の事実により、onEvent 呼び出し元スレッドがそのメソッドから返されず、onEvent() 内で無限に待機しているように見えるためです。

現時点では、その呼び出しのタイムアウトを作成することはできません。

質問は::

ディスラプター エンジンが onEvent() を呼び出すとき、バッファ作成レベルまたはその他のレベルでタイムアウトを設定できますか。onEvent() メソッド呼び出しでタイムアウトが発生します。

BlockingWaitStrategy でディスラプター 3.3.2 を使用しています。