問題タブ [blockingqueue]

キュー インターフェイスを実装するカスタム キューを作成しています。この実装はスレッド セーフであり、場合によってはブロックされます。

InterruptedException通常の Queue インターフェースは例外について言及していないため、実装で例外をスローすることはできません。

この問題には 2 つの解決策がありますが、どちらも満足できるものではありません。

1. Queue インターフェイスを削除し、例外をスローします。これにより、Queue を必要とする外部ソフトウェアでコードを使用できなくなります。

2. スローRuntimeException、これにより、非常に驚​​くべきソフトウェアアクティビティが大量に発生しますが、リスクを冒したくありません。

どういうわけか、実装のようなArrayBlockingQueue実装QueueとBlockingQueue. それは行くべき道ですか、それともここでのトリックは何ですか?

インターフェイスJUnitをテストするための既存のテストは既にありますか? BlockingQueueダウンロードできるいくつかのクラスは、再生を押すと、すべてのテストケースを考えるのに1日を費やすことなく、赤に変わります(うまくいけば緑:-))?

私はグーグルで検索しましたが、何も見つかりませんでした。

いくつかのアイデア？

編集: G_Hが提供したテスト ケースは、基本的なデバッグに非常に適していました。マルチスレッドを頻繁に使用するための適切なテストがまだありますか?それは実際に競合やデッドロックを引き起こしますか?

私にはブロッキングキューがあり、ワーカーはそのキューから取得して、約1〜10ミリ秒間作業しています。

特定の時間枠でワーカーがアクティブでアイドル状態だった時間の相対的な数値を把握したいと思います（たとえば、直前に70％がアクティブでした）。

短い1msの期間を合計すると、不正確になる可能性があり、nano時間をフェッチすると多くのパフォーマンスが消費されます。

そのようなタスクを実行するためのいくつかのきちんとした方法はありますか？

これらのキューワーカーの設定はコードで頻繁に行われるため、実行時にボトルネックのプロファイルを作成できるようにしたいと思います。

編集：スレッドが機能しているかどうかだけを知りたいです。OSがスレッドを処理フォーカスから外した場合でも、カウントされます。キューロックを待機している場合のみ、アイドル時間としてカウントする必要があります。

BlockingQueueからタスクをプルする通常のスレッドを作成する代わりに、Android Looper機能を使用して「パイプラインスレッド」を作成する必要がある理由を誰かが説明できますか？表面的には、同じことを行うには2つの方法のように見えます。

違いはなんですか？キャパシティ1でいつ使用する必要がありますSynchronousQueueか？LinkedBlockingQueue

私は現在、Bluetooth接続を介して着信データのバッファとして使用するクラスを実装しようとしています:

このバッファに問題があります。バイト配列を追加するたびに、バッファ内のすべてのバイト配列が追加したものと同じになります。

コードの残りの部分で (独自のクラスにせずに) 同じタイプのブロッキング キューを使用してみましたが、正常に動作します。問題は、このクラスを使用するときのようです。

クラスを宣言する方法は次のとおりです。

私が間違いを犯している場所を誰かが見ることができますか?

私はこれに本当に困惑しており、過去3日間デバッグしようとしました。うまくいけば、誰かが私が間違っていることを教えてくれるでしょう。

Bluetooth 経由で PC からストリーミングされる情報を受信するために、BlockingQueue (FIFO) バッファーを実装しています。RealTerm を使用してハイパーターミナル リンクを介して事前に記録された心電図信号を送信しています。

値を追加してから削除してアプリを起動するときにバッファをテストしましたが、正常に機能しているようです。

Bluetooth接続からデータを受信して​​いるときにバッファに保存しようとすると、問題が発生します。BlockingQueue が処理できるよりも速く追加しているかどうかはわかりませんが、データ送信を停止してバッファを確認すると、バッファ全体に最後に追加された値が含まれています。バッファのサイズは正しいが、内容が正しくありません。

ここに私のバッファがあります:

情報を受信して​​バッファに送信する BluetoothCommunication クラスの部分は次のとおりです。

したがって、私の問題の例は次のようになります。

Bluetooth 接続を介して値を送信します (1 から 20 までの 8 ビット値)。IncomingBuffer クラスの挿入メソッドでは、適切な値が送信されたことをログ メッセージが確認します。値がバッファから取得されると、最後に挿入された数値 (20) をすべて含む 20 個のバイト配列が含まれます。

バッファが他の状況では機能するが、Bluetooth 通信中は機能しない理由についての手がかりはありますか?

2つのブロッキングキューがある状況があります。最初に、実行するタスクをいくつか挿入します。各タスクが完了すると、タスクが2番目のキューに追加され、そこで実行されます。

したがって、最初のキューは簡単です。空でないことを確認して実行するだけです。それ以外の場合は、interrupt（）を実行します。

あなたが言うことができるように、私が次のことをする2番目のものは機能しません：

2番目のBlockingQueueがtake（）でブロックせず、追加するアイテムがこれ以上ないことがわかった場合にのみ終了するように、どのように解決しますか。2番目のスレッドがおそらく1番目のブロッキングキューを認識し、それが空であり、2番目のキューも空であるかどうかを確認すると、割り込みが発生する可能性があります。

Poisonオブジェクトを使用することもできますが、他のものを使用したいと思います。

注意：これは正確なコードではなく、私がここに書いたものです。

私は、pthreads、semaphore.h、および gcc アトミック ビルトインの上にある循環バッファーに支えられた高性能ブロッキング キューを実装しようとしています。キューは、異なるスレッドからの複数の同時読み取りおよび書き込みを処理する必要があります。

ある種の競合状態を分離しましたが、それがアトミック操作とセマフォの動作に関する誤った仮定なのか、それとも私の設計に根本的な欠陥があるのか​​ はわかりません。

それを抽出して、以下のスタンドアロンの例に単純化しました。このプログラムが二度と戻ってこないことを期待しています。ただし、キューで破損が検出された数十万回の反復後に戻ります。

以下の例 (解説用) では、実際には何も格納されず、実際のデータを保持するセルを 1 に設定し、空のセルを表すために 0 を設定します。空いているセルの数を表すカウンティング セマフォ (Vacancy) と、占有されているセルの数を表すカウンティング セマフォ (Occupants) があります。

ライターは次のことを行います。

1. 空室を減らす
2. アトミックに次のヘッド インデックスを取得します (mod キュー サイズ)
3. それに書き込みます
4. 占有者を増やす

読者は反対のことをします:

1. 占有者を減らす
2. 次のテール インデックスをアトミックに取得する (mod キュー サイズ)
3. そこから読む
4. 空室を増やす

上記を考えると、正確に1つのスレッドが特定のセルを一度に読み書きできると思います。

機能しない理由やデバッグ戦略についてのアイデアを歓迎します。以下のコードと出力...

上記を次のようにコンパイルします。

出力は毎回異なりますが、一例を次に示します。

私のシステムは Intel Core 2 の Ubuntu 11.10 です:

ありがとう、アンドリュー。

いくつかのタスクオブジェクトを作成するシングルスレッドプロデューサーがあり、ArrayBlockingQueueそれらは（固定サイズの）に追加されます。

また、マルチスレッドのコンシューマーを開始します。これは、固定スレッドプール（Executors.newFixedThreadPool(threadCount);）としてビルドされます。次に、いくつかのConsumerWorkerインスタンスをこのthreadPoolに送信します。各ConsumerWorkerは、上記のArrayBlockingQueueインスタンスを参照します。

そのような各ワーカーはtake()キューで実行し、タスクを処理します。

私の問題は、これ以上行うべき作業がなくなるときに、ワーカーに知らせるための最良の方法は何であるかということです。つまり、プロデューサーがキューへの追加を終了したことをワーカーに通知するにはどうすればよいですか。この時点から、各ワーカーはキューが空であることを確認したら停止する必要があります。

私が今持っているのは、私のプロデューサーが（キューに何かを追加するという）仕事を終えたときにトリガーされるコールバックで初期化されるセットアップです。また、作成してThreadPoolに送信したすべてのConsumerWorkerのリストも保持しています。プロデューサーコールバックがプロデューサーが完了したことを私に告げるとき、私はこれを各労働者に伝えることができます。この時点で、キューが空でないかどうかを確認し続ける必要があります。キューが空になると停止するため、ExecutorServiceスレッドプールを正常にシャットダウンできます。こんな感じです

}

ここでの問題は、プロデューサーが終了してシグナルを送信し、ConsumerWorkersがキュー内のすべてを消費する前に停止するという明らかな競合状態があることです。

私の質問は、これを同期してすべてが正常に機能するようにするための最良の方法は何ですか？プロデューサーが実行されているかどうかをチェックし、キューが空であるかどうかをチェックし、さらに1つのブロック（キューオブジェクト上）のキューから何かを取得する部分全体を同期する必要がありますか？isRunningConsumerWorkerインスタンスのブール値の更新を同期する必要がありますか？他に何か提案はありますか？

更新、これが私が使用することになった実用的な実装です：

}

これは、いくつかの小さな変更を加えただけで、以下のJohnVintの回答に大きく影響を受けました。

===@vendhanのコメントによる更新。

どうぞよろしくお願いいたします。そうです、この質問の最初のコードスニペットには（他の問題の中でも）while(isRunning || !inputQueue.isEmpty())実際には意味がないものがあります。

これの実際の最終的な実装では、「||」を置き換えるというあなたの提案に近いことをします。（または）「&&」（および）を使用すると、各ワーカー（消費者）は、リストから取得した要素が毒薬であるかどうかのみをチェックし、停止した場合は停止します（理論的には、ワーカーは実行中であり、キューが空であってはなりません）。