問題タブ [phaser]

ExecutorServiceプロデューサースレッドがN個の作業項目を生成し、それらをに送信してから、 N個の項目がすべて処理されるまで待機する必要があるという問題が2回発生しました。

警告

• Nは事前にはわかりません。もしそうなら、私は単にを作成し、すべての作業が完了するまでCountDownLatchプロデューサースレッドを作成します。await()
• 私のプロデューサースレッドはブロックする必要がありますが（つまり、を呼び出すことによって）、プロデューサースレッドの待機を停止させるために、すべての作業が完了したことを通知する方法がないため、を使用することCompletionServiceは不適切です。take()

私の現在の好ましい解決策は、整数カウンターを使用し、作業項目が送信されるたびにこれをインクリメントし、作業項目が処理されるときにそれをデクリメントすることです。N個のタスクすべてを送信した後、プロデューサースレッドはロックを待機し、counter == 0通知されるたびにチェックする必要があります。コンシューマースレッドは、カウンターがデクリメントされ、新しい値が0の場合、プロデューサーに通知する必要があります。

この問題へのより良いアプローチはありjava.util.concurrentますか、それとも「自分でローリング」するのではなく、使用すべき適切な構成がありますか？

前もって感謝します。

この質問は2年前に尋ねられましたが、言及されているリソースがあまり役に立たない（IMHO）か、リンクが無効になっています。

理解するには、いくつかの優れたチュートリアルが必要Phaserです。javadocを読みましたが、目が眩むほどです。javadocを本当に理解するには、これらのクラスがどのように使用されるかを知っている必要があるからです。

誰か提案がありますか？

与えられた：

doSomething()タスクの内部状態に応じて、自分が最後のタスクであるかどうかを確認したいと思います。Phaser.onAdvance（int、int）を見つけましたが、この場合の使用方法が明確ではありません。

それで、私は非常に柔軟なこの Phaser を手に入れましたが、何かが足りないようです。
私は CyclicBarrier の使用に成功しましたが、今はもっと柔軟なものが必要です。コードは次のとおりです

。

コード：

sp.getSomething();I/O 呼び出しです。
このコード サンプルが複数のスレッドによって実行されていることを考慮してください。

ここに私の問題があります: MAX_CLIENTS 以上が Phaser に到着しないことを確認したので、MAX_CLIENTS が到着した場合はすべて問題ありません。ただし、TimeoutException で問題が発生しました。1 つ目は、クライアント (スレッド A など) がフェーズに到達できるタイム ウィンドウ (競合状態) であり、次にスレッド B で TimeoutException が発生します。スレッド B で別のフェイザーを到着したパーティの数で動的にインスタンス化しています。 (たとえば 5)、スレッド A は既にフェーズに到達しています (別名、phase1a は < 0 であることがわかりませんでした)。どうすればそれを修正できますか? 私はセマフォを使用することを考えていましたが、おそらくこれを行う方法を再考する必要があるため、努力する価値はないと思います. タイマーを使用してインクリメントすることも考えましたAtomicInteger変数とタイマーが切れると、Phaser が動的にインスタンス化されます。この問題にどのようにアプローチするかについてのアイデアはありますか?

編集:
ドキュメントにはbulkRegister(int parties)メソッドがありますが、ちょっと変わった言い回しです:

指定された数の新しい未到着のパーティをこのフェイザーに追加します。onAdvance(int, int) の継続的な呼び出しが進行中の場合、このメソッドはその完了を待ってから戻る場合があります。このフェーザーに親があり、指定されたパーティーの数が 0 よりも大きく、このフェーザーに以前にパーティーが登録されていなかった場合、この子フェーザーもその親に登録されます。このフェーザーが終了した場合、登録しようとしても効果がなく、負の値が返されます。

質問:「かもしれません」という言葉は私を混乱させます! "may" as may または "may"

編集：
解決しました。以下の私の答えを確認してください。

Phaser を使用していつ作業を終了するかを知るマルチスレッド アプリケーションを作成しています。問題は、ExecutorCompletionService ではキューに 100k のスレッドが存在する可能性があることですが、Phaser での未到着パーティの最大数は 65535 です。65536 パーティが到着した場合、どうすればよいですか?

私のコード例:

問題は addParties() メソッドにあります。シングル スレッド (SimpleParser) は 100 個の新しいファイル名を返すことができ、100 個の新しいスレッドが ExecutorCompletionService に送信され、100 個の新しいパーティが Phaser に登録されます。私はこのようなものを使用しようとしました:

フェイザーのチェーンを作成しますが、p.getUnarrivedParties（）が0を返すため、役に立ちませんでしたが、次のパーティーを登録できません...

プリント:

65535

0

IllegalStateException をスローします

では、この古いものに接続される新しい Phaser を作成するにはどうすればよいでしょうか?

//編集

ありがとう@ボウモア。あと2つだけ質問があります。

例を見てみましょう：

それは印刷します：

なぜ後 child1.arriveAndDeregister(); child1 は終了していませんが、実際に終了しているかどうかを確認する方法は?

2 番目の質問です。何千もの新しいオブジェクトを作成するのは無意味だと思ったので、65535 パーティーに達した後、新しい Phaser を作成することについて尋ねました。これでメモリの問題は発生しないと思いますか、それともパフォーマンスを向上させることさえできると思いますか?

少数のスレッドを生成し、実行時にリストに配置しようとしています。彼らが処理を完了したら、プレゼンテーションのために結果を収集したいと思います。そうすれば、多くのスレッドを含むリストを作成し、それらが利用可能になったらfuture.get、それらのコールバック情報を呼び出して使用できます。

何らかの理由で、結果の多くが欠落しています。コードをステップ実行すると、f.get()渡されるべきではないときに渡され、その理由がわかりません。

私のコードは次のとおりです。

なぜこれが起こっているのかについてのアイデアはありますか？

編集：

さて、私はそれを稼働させていると思います：

Phaser そこについてのjavadocを読んだところですが、そのクラスの使用法について質問があります。javadoc はサンプルを提供しますが、実際の例はどうでしょうか? そのようなバリアの実装が実際に役立つのはどこですか?