問題タブ [blockingqueue]

JavaDoc for ThreadPoolExecutorBlockingQueueは、エグゼキュータのバッキングにタスクを直接追加できるかどうかについては不明です。ドキュメントによると、呼び出しexecutor.getQueue()は「主にデバッグと監視を目的としています」。

私はThreadPoolExecutor自分でを構築していBlockingQueueます。キューへの参照を保持しているので、タスクを直接キューに追加できます。同じキューが返されるgetQueue()ので、の警告はgetQueue()、私の手段で取得したバッキングキューへの参照に適用されると思います。

例

コードの一般的なパターンは次のとおりです。

queue.offer()vsexecutor.execute()

私が理解しているように、一般的な使用法は、を介してタスクを追加することですexecutor.execute()。上記の私の例のアプローチには、キューをブロックするという利点がありexecute()ますが、キューがいっぱいになり、タスクが拒否されるとすぐに失敗します。また、ジョブの送信がブロッキングキューと相互作用することも気に入っています。これは私にとってより「純粋な」生産者/消費者のように感じます。

キューにタスクを直接追加することの意味：呼び出す必要がありますprestartAllCoreThreads()。そうしないと、ワーカースレッドが実行されません。エグゼキュータと他の相互作用がないと仮定すると、キューを監視するものはありません（ThreadPoolExecutorソースの調査によりこれが確認されます）。これは、直接エンキューのThreadPoolExecutor場合、0を超えるコアスレッド用に追加で構成する必要があり、コアスレッドがタイムアウトできるように構成してはならないことも意味します。

tl; dr

次のようにThreadPoolExecutor構成されている場合：

• コアスレッド>0
• コアスレッドはタイムアウトできません
• コアスレッドは事前に開始されています
• BlockingQueue遺言執行者の支援への参照を保持する

呼び出す代わりに、タスクをキューに直接追加することはできますexecutor.execute()か？

関連している

この質問（プロデューサー/コンシューマーワークキュー）も同様ですが、キューに直接追加することについては特に説明していません。

Java プロセスを実行しているときに、Linux (64 ビット) への最近の顧客の移行に問題があります。

このプロセスは、ほとんどが futex の識別子を持つ何千ものスレッドを生成しています。私は futex (高速ユーザー空間ミューテックス) を調べましたが、これは基本的なロックを実装するための Linux 構造です。

コードは最近、BlockingQueue と ExecutorService を実装して子スレッドを生成するように変更されましたが、子スレッドの数は構成設定によって制御されており、この特定のメカニズムが暴走していないことを証明できます。BlockingQueue といくつかのロックのために、JVM 内部の何かがこれらすべてのスレッドを生成しているとしか思えませんか?

それで、これらのスレッドが実際に何であるかを知る方法と、それらを制御/停止するために私ができることを誰か教えてもらえますか?

以下は、プロセス リストの数行です。プロセスを強制終了する前の実際のリストは、13000 行を超えていました。

0 - 54321 447 446 1 - - - 5953085 - ? 00:15:50 Java
0 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -
1 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -
1 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -
1 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -
1 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -
1 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -
1 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -
1 S 54321 - - 0 82 2 - - futex_ - 00:00:00 -

どんな提案もありがたく受け入れました。

私はこのコードを持っています。Aは、キューへの追加を待機しているときに中断された場合にLinkedBlockingQueueのみ、をスローする必要があります。Exceptionただし、このキューには制限がないため、できるだけ早く追加する必要があります。シャットダウンメソッドがスローするのはなぜInterruptedExceptionですか？

私は、ブロッキング キューとロックフリー キューを認識しています。これは、スコットらによって提供されている実装の好例です。、しかし、ロックフリーブロッキングキューの実装はありますか?

lock-free-blocking-queue では、dequeue はロックを必要としませんが、キューにアイテムがない場合、コンシューマーをブロックします。そのような獣の実装はありますか? それらが C# 実装であることが望ましいですが、どの実装でも技術的には機能します。

アップデート：

D14.1 行で競合状態になると思います。

Java マルチスレッド構造を理解しようとしています。また、ブロッキング キューの簡単な実装を書こうとしています。ここに私が書いたコードがあります：

Take() で待機しているスレッドに put() から、またはその逆に通知したい。誰かがこれを行う正しい方法を教えてください。

java.utils の実装を確認したところ、この段階では少し複雑な Condition と ReentrantLocks が使用されています。今のところ、簡単にするために、完全に堅牢でなくても大丈夫です[しかし正しい].

ありがとう ！

http://download.oracle.com/javase/1.5.0/docs/api/java/util/concurrent/BlockingQueue.html#putの「典型的な生産者と消費者のシナリオに基づく使用例」を見ていました。 (エ)

例は正しいですか？

put および take 操作では、キューの変更に進む前に何らかのリソースをロックする必要があると思いますが、ここではそれが行われていません。

また、これが並行キューの種類であった場合、並行キューでのアトミック操作はロックを必要としないため、ロックの欠如は理解できたでしょう。

そのため、プロデューサー/コンシューマー タイプのアプリケーションで固定サイズの BlockingQueue [ArrayBlockingQueue] を使用していますが、ユーザーがその場でキュー サイズを変更できるようにしたいと考えています。問題は、作成後に容量を変更できる BlockingQueue 実装がないことです。誰もこれに遭遇したことがありますか？何か案は？

BlockingQueueのドキュメントによると、一括操作はスレッドセーフではありませんが、drainTo() メソッドについては明示的に言及されていません。

BlockingQueue の実装はスレッドセーフです。すべてのキューイング メソッドは、内部ロックまたはその他の形式の同時実行制御を使用して、アトミックに効果を達成します。ただし、一括コレクション操作の addAll、containsAll、retainAll、および removeAll は、実装で特に指定されていない限り、必ずしもアトミックに実行されるとは限りません。そのため、たとえば、c の一部の要素のみを追加した後、addAll(c) が失敗する (例外がスローされる) 可能性があります。

drainTo() メソッドのドキュメントでは、BlockingQueue の要素が排出されるコレクションは、スレッドセーフな方法で変更できないことが指定されています。しかし、drainTo() 操作がスレッドセーフであることについては何も言及されていません。

このキューから使用可能なすべての要素を削除し、指定されたコレクションに追加します。この操作は、このキューを繰り返しポーリングするよりも効率的です。コレクション c に要素を追加しようとして失敗した場合、関連付けられた例外がスローされたときに、要素がどちらのコレクションにも含まれないか、いずれかのコレクションに含まれないか、または両方のコレクションに含まれない可能性があります。キューをそれ自体にドレインしようとすると、IllegalArgumentException が発生します。さらに、操作の進行中に指定されたコレクションが変更された場合、この操作の動作は未定義です。

では、drainTo() メソッドはスレッドセーフですか? 言い換えると、あるスレッドがブロッキング キューで drainTo() メソッドを呼び出し、別のスレッドが同じキューで add() または put() を呼び出している場合、両方の操作の終了時にキューの状態は一貫していますか?

それは私と私のBlockingQueueです...私はこの記事と この質問に従ってそれを書き直しました。いくつかのアイテムを送信した後、アクセス違反でクラッシュします。コードは次のとおりです。

Pop（）の呼び出しで、ifの後の行で常にクラッシュします。

これは、cosのpNewHeadNodeがNULLであることを示しています。しかし、これはどのように起こりますか？

編集：初期化コードを忘れた：