問題タブ [blockingqueue]

この実装では要素が追加されていないため、ファイルのコンテンツからキューに入力する場合、深さは増加しないようです。

offerで置き換えるadd場合、例外がスローされた場合

私は何を間違っていますか？最初の要素を挿入するとすぐにキューがいっぱいになるのはなぜですか?

テキストファイルを読み取ってから変換し、他のファイルに書き込む必要があるという要件があります。読み取り用のスレッド、変換用のスレッド、書き込み用のスレッドのように、これを並行して実行したいと考えています。
スレッド間でデータを共有するには、いくつかのチャネルが必要です。これに使用することを考えていBlockingQueueましたが、利用可能な場合は他の (より良い) 代替手段を検討したいと考えています。
Guava にはEventBus要件がありますが、これが要件に適しているかどうかはわかりません。他にどのような選択肢があり、パフォーマンスの観点からどれが最適か。

In java, there is nice package java.util.concurrent which holds implementation for BlockingQueue interface.

I need something similar in Haskell, so it will be able to

• maintain fixed size of queue in memory
• block read operations when queue is empty (get)
• provide time-boxed blocks, which will return Nothing if queue is empty and timeout exceeded
• similar for put operations - block until queue has capacity with time-boxed version

probably this could be implemented with STM or blocking transactions - but I was not able to find something like that on hackage.

キュー内のオブジェクトをバッチ処理できる必要があることを除いて、JavaBlockingQueueと同じデータ構造に興味があります。つまり、プロデューサーがオブジェクトをキューに入れることができるようにしたいのですが、take()キューが特定のサイズ（バッチサイズ）に達するまでコンシューマーブロックをオンにします。

次に、キューがバッチサイズに達すると、プロデューサーはput()、コンシューマーがキュー内のすべての要素を消費するまでブロックする必要があります（この場合、プロデューサーは再び生産を開始し、コンシューマーはバッチに再び達するまでブロックします）。

同様のデータ構造が存在しますか？または、それを書く必要があります（私は気にしません）、何かがそこにある場合、私は自分の時間を無駄にしたくありません。

アップデート

多分少し物事を明確にするために：

状況は常に次のようになります。複数のプロデューサーがキューにアイテムを追加することもできますが、キューからアイテムを取得するコンシューマーが複数存在することはありません。

さて、問題は、これらのセットアップが並列および直列に複数あることです。つまり、プロデューサーは複数のキューのアイテムを作成しますが、コンシューマー自体もプロデューサーになることができます。これは、生産者、消費者-生産者、そして最終的には消費者の有向グラフとしてより簡単に考えることができます。

キューが空になるまでプロデューサーがブロックする必要がある理由（@Peter Lawrey）は、これらのそれぞれがスレッドで実行されるためです。スペースが利用可能になったときに単純に生成するためにそれらを残すと、一度に処理しようとするスレッドが多すぎるという状況になります。

たぶんこれを実行サービスと組み合わせると問題が解決するでしょうか？

ブロッキングを使用したプロデューサー/コンシューマーの実装があり、次のように開始します

Producer内部はこんな感じ

コンシューマーは次のようになります

しかし、しばらくすると非常に遅くなりますが、とにかく非常に高速に維持できますか?

次のblockingQueueがあります。

どこ

メッセージをキューに入れるプロデューサーがあります。

今私の目標は、特定の作成者の特定のメッセージのみをキューからフェッチできるコンシューマーを作成することですか? 出来ますか？

そうでない場合、BlockingQueue に代わるものは何ですか?

最近、BlockingQueue のエンキューの次の実装を見ました ( source )

whileループが必要な理由と、ループwhileを置き換えることができる理由if (this.queue.size() == this.limit)

メソッドのエンキューが同期されているように見えるため、一度に 1 つのスレッドのみがメソッド本体で実行され、wait(). this.queue.size() == this.limitスレッドが通知されたら、再度状態を確認せずに先に進むことはできませんか?

ArrayBlockingQueue を使用するワーカーがいくつかあります。

すべてのワーカーはキューから 1 つのオブジェクトを取得して処理し、その結果、さらに処理するためにキューに入れられる複数のオブジェクトを取得できます。つまり、労働者＝生産者＋消費者です。

ワーカー：

主要：

仕事がなくなったときにワーカーを停止する最良の方法は何ですか?

まず、私が念頭に置いているのは、キューにあるアイテムの数と現在処理中のアイテムの数を定期的に確認することです。両方がゼロの場合は、ExecutorService で「shutdownNow()」などを実行します。しかし、これが最善の方法であるかどうかはわかりません。

私は次のコードを書きましたが、どこかで間違っていると感じています。

ここで、add（String）は文字列をキューに追加します。UDPポートから入力を受信するたびに呼び出されます。process（）はキューを処理し、処理のためにメッセージクラスに送信します。出力されたロックおよび解放された印刷ステートメントが目的の順序ではありません。

編集

私の予想される答えは次のようになります。プロデューサーに追加された場合は、プロデューサーによってロックされます->キューに追加->リリースをロックします。同じように消費者にもなります。ただし、操作はインターリーブしないでください。つまり、プロデューサーによってロックされたものが印刷されると、コンシューマーによってロックされた状態で印刷されてからロックが解除されないようにする必要があります。

ExecutorService と ArrayBlockingQueue を使用して、プロデューサーとコンシューマーの設計に関する私の理解が正しいかどうかを知りたいです。この設計を実装するにはさまざまな方法があることは理解していますが、最終的には問題自体に依存すると思います。

私が直面しなければならなかった問題は次のとおりです。大きなファイル (6 GB) から読み取る 1 つのプロデューサーがいます。行ごとに読み取り、すべての行をオブジェクトに変換します。オブジェクトを ArrayBlockingQueue に配置します。

コンシューマー (少数) は ArrayBlockingQueue からオブジェクトを取得し、これをデータベースに永続化します。

さて、明らかに生産者は消費者よりもはるかに高速です。各行をオブジェクトに変換するのに数秒かかりますが、消費者にとってはより長い時間がかかります。

したがって...これを行うことでこのプロセスをスピードアップしたい場合: 2 つのクラス化された「ProducerThread」と「ConsumerThread」を作成し、それらは ArrayBlockingQueue を共有します。2 つの間で調整するスレッドは次のようになります。

私の質問は次のとおりです。

1. 上記の設計は実際に各スレッドを同時に使用しますか? 私のコンピューターは 2.4GHz クアッドコア 2 台です。

2. Future と .get() が何のためにあるのかわかりませんか?

ちなみに、結果は高速です (最初のバージョンはシーケンシャルで、3 時間かかったと考えてください)。今では 40 分ほどかかります (ただし、改善の余地があるかもしれません)。

ポインタをありがとう