問題タブ [blockingqueue]

私は、C++ で stl キューに同時追加すると問題が発生する可能性があることを知っています。これを解決する方法は、すべての追加/削除呼び出しにミューテックス ロックを追加することです。

しかし、私は現在 Java でプログラミングしており、BlockingQueue を使用しています。ドキュメントには、BlockingQueue オブジェクトでプット/テイクを呼び出すスレッドは、それぞれプット/テイクする余地ができるまで暗黙的にブロックされるとのみ記載されています。ただし、同時の put/take 呼び出しについては何も言及されていません。これらをミューテックスロックで保護する必要がありますか?

オブジェクト B と C の何百ものインスタンスを作成するアプリケーションがあります。

オブジェクト Foo に 2 つのキュー (b_queue と c_queue) が含まれるオブジェクト階層があり、一方にはタイプ B のオブジェクトが格納され、もう一方にはタイプ C のオブジェクトが格納されます。

BまたはCで何らかのイベントが発生したときに、オブジェクトFooのキューから自分自身を削除してもらいたい.

私の解決策は、オブジェクト B または C を作成して、オブジェクトをキューに格納する Foo のインスタンス (foo_inst と呼びます) を渡すことです。次に、B または CI 内で foo_inst.b_queue.remove(this) を呼び出すだけです。

1）これは良い/効率的な方法ですか、それとも避けるべきですか？

B と C はどちらも Runnable オブジェクトであり、ThreadPoolExecutor でキューに入れられます。これは、それらがワークキューに 2 回存在する可能性があり、同時に foo_inst.b_queue.remove(this) を呼び出そうとするか、または既に実行された後に呼び出そうとする可能性があることを意味します。

2) これも問題になりますか?

ヘルプやヒントは大歓迎です。

さて、これまでのところ、いくつかの助けを借りてここまでたどり着くことができました（これ以上の助けを歓迎します）：

}

私は次のような状況にあります:

1. データベースからデータを読み取る
2. 仕事「計算」をする
3. 結果をデータベースに書き込む

データベースから読み取り、生成されたオブジェクトを BlockingQueue に入れるスレッドがあります。これらのオブジェクトは非常に重いため、キューはメモリ内のオブジェクトの量を制限します。複数のスレッドがキューからオブジェクトを取得し、作業を実行して、結果を 2 番目のキューに入れます。最後のスレッドは、2 番目のキューから結果を取得し、結果をデータベースに保存します。

問題は、デッドロックを防ぐ方法です。「計算スレッド」は、オブジェクトがキューに入れられなくなる時期を知る必要があります。現在、スレッド (呼び出し可能) の参照を相互に渡し、ポーリングまたはオファーの前に thread.isDone() をチェックし、要素が null であるかどうかを確認することで、これを実現しています。キューのサイズもチェックします。要素が含まれている限り、キューを消費する必要があります。take または put を使用すると、デッドロックが発生します。

これを達成するためのより簡単な方法はありますか？

誰かがアイテムを渡したObjectときに、非同期で通知を受ける必要があります。BlockingQueue

Javadoc と Web の両方で事前に作成されたソリューションを検索した後、最終的に私の (おそらくナイーブな) ソリューションになりました。

と

基本的に、操作が成功するたびtake()にリスナーオブジェクトをコールバックするキューの操作をブロックするスレッドがありtake()、オプションで特別なオブジェクトを送り返しcookieます（必要に応じて無視してください）。

質問は次のとおりです。これを行うためのより良い方法はありますか? (同時実行性/効率性および/またはコードのクリーンさの両方で) 許されざる間違いを犯していますか? 前もって感謝します。

オブザーバーパターンとBlockingQueueを使用してインスタンスを追加しています。別のメソッドでキューを使用していますが、次のように実行しているにもかかわらず、take（）が永久に待機しているようです。

一方、mDiffQueueはLinkedBlockingQueueであり、次のように使用しています。

しかし、mDoneがtrueではないのに最初の式がチェックされていると思います。その後、mDoneがtrueに設定されている可能性があります（オブザーバーは常にマルチスレッドですか？）が、すでにmDiffQueue.take（）を呼び出していますか？：-/

編集：私は本当に今それを取得していません。最近、次のように変更しました。

デバッガーで少し待つと動作しますが、mDoneがfalseに初期化され、while条件がtrueになり、本体が実行されるため、「リアルタイム」でも動作するはずです。

mDiffQueueが空で、mDoneがtrueの場合、whileループの本体をスキップする必要があります（つまり、キューはもういっぱいになりません）。

編集：それはそうです：

なぜpeek（）が必須なのかわかりませんが。

編集：

私がしているのはツリーを反復処理することであり、すべてのINSERTEDノードをスキップしたいと思います。

基本的に、ツリーの別のリビジョンで削除されたノードを考慮せずに、ツリーの最大深度またはレベルを計算します（サンバーストの視覚化の比較のため）が、それは範囲外である可能性があります。最大深度を調整しているだけでも、挿入されていないノードで何かをしていることを説明するためだけに。

よろしく、

ヨハネス

パイプラインアプローチを実装しました。ツリーをトラバースするつもりですが、事前に利用できない特定の値が必要です...したがって、ツリーを並行して（または前に）トラバースし、値を保存したいノードごとにもう一度トラバースする必要があります（たとえば、descendantCount ）。

そのため、ツリーを介して対話し、コンストラクターから、ExecutorService を通じて開始された新しいスレッドを呼び出すメソッドを呼び出しています。送信される Callable は次のとおりです。

したがって、すべてのノードに対して、すべてのノードのツリーをトラバースし、子孫と変更をカウントする新しい Callable を作成しています (実際には、2 つのツリー リビジョンを融合しています)。さて、mDescendantsQueue と mModificationQueue は BlockingQueue です。最初は、descendantsQueue しか持っておらず、ツリーをもう一度トラバースして、すべてのノードの変更を取得しました (現在のノードのサブツリーで行われた変更を数えます)。次に、両方を並行して実行し、パイプライン化されたアプローチを実装しない理由を考えました。悲しいことに、別のマルチスレッドの「ステップ」を実装するたびに、パフォーマンスが低下したように見えました。

おそらく、XML ツリーは通常それほど深くなく、同時実行オーバーヘッドが重すぎるためです :-/

最初はすべてを順番に実行しましたが、これが最も高速でした:

BlockingQueues でパイプライン化されたアプローチを使用した後、パフォーマンスが低下したように見えますが、実際には時間測定を行っていないため、多くの変更を元に戻して元に戻す必要があります:( CPU が増えるとパフォーマンスが向上する可能性があります。今すぐテストする Core2Duo。

よろしく、
ヨハネス

PriorityQueue に関する同様の質問を見つけました。そのエラーは、正しく初期化されていないというものでした。私は同じ問題を抱えているかもしれませんが、正しく初期化する方法がわかりません!

今のところ、私はただやっています:

BlockingQueue myQueue = null;

しかし、リストに何かを追加しようとするとすぐに例外がスローされます。

BlockingQueue を正しく初期化するにはどうすればよいですか?

メソッドBlockingQueueを使用してスレッドが要素を吸い込むと、要素が利用可能になるまで待機することを理解しています（中断されない限り）。take()

2 つの質問があります。

i) 要素が使用可能になるとすぐに、スレッドは自動的に起動されますか、それとも遅延がありますか (つまり、スレッドは後でそれ自体をチェックします)?

ii) 遅延がある場合、スレッドをウェイクアップすることは意味がありますか (たとえば、明示的に中断することによって)? レイテンシとパフォーマンスについて考えています。

LinkedBlockingQueue が必要ですが、それにプリミティブを渡しています。キューに追加するための私のデータ レートは、約 4 ミリ秒または毎秒 256 データ ポイントです。私が抱えている問題は、データが開始時にすぐに遅延し始めることですが、時間が経つにつれてJITがこれをより効率的にし、リアルタイムで終了するようです。初期遅延を減らす必要がある場所を見つけようとしていますが、そのうちの 1 つは、キューへの挿入ごとにオートボクシングから Float オブジェクトを「新しく」作成することです。プリミティブを使用して LinkedBlockingQueue を持っている人はいますか?

または、サイズやプリミティブの使用がわからない場合、 LinkedBlockingQueue よりも高速なものはありますか?

特に、ブロッキング キューを探しています。C++11にそんなものあるの？そうでない場合、他にどのような選択肢がありますか? もう自分でスレッドレベルに行きたくありません。あまりにもエラーが発生しやすい。