問題タブ [executorservice]

ExecutorServiceオブジェクトを実行しようとしたところFixedThreadPool、問題が発生しました。

プログラムがナノ秒で実行されることを期待していましたが、ハングしました。Semaphoreキュー内のアイテムが合計されないように、それと一緒に使用する必要があることがわかりました。

プールのすべてのスレッドが使用されていることを知る方法はありますか?

基本コード...

このコードはハングアップしますが、その理由は本当にわかりません。プールが現在すべてのスレッドが終了するのを待っているかどうかを知る方法は?

多数のメッセージを処理する必要があるという要件があります。これらのメッセージは、別のプロセスによって DB テーブルに挿入されます。私がしなければならないのは、DB で新しいメッセージをチェックし、構成に応じて対応する顧客に電子メールまたは http で送信することだけです。メッセージの数は常に数千に達する可能性があり、顧客の数は約 1000 です。

生産者と消費者が機能する方法でこれを設計する予定です。プロデューサー スレッドが DB をポーリングして新しいメッセージを探し、それらを Queue に入れるように、ワーカー スレッドがこれらのメッセージとプロセスを読み取ります。

最初は、JMS がこの要件に適したソリューションであるように見えました。ただし、次のシナリオを考えると、この要件に適した Threadpool を使用して、ExecutorService のような代替手段があるかどうかを探しています。

1. 1 つのメッセージ配信が失敗した場合、少なくとも 24 時間は何度か再試行する必要があります。
2. 顧客に対して 1 つのメッセージ配信が失敗した場合、他のメッセージ配信もすべて失敗する可能性があります。したがって、その顧客への次のメッセージを処理する前に、最初のメッセージを送信する必要があります。

つまり、すべての顧客のすべてのメッセージに対して 1 つのキューがある場合、1 つのメッセージが失敗すると、他のメッセージは処理されません。

これをどのように処理するのが最善かについて誰かが提案してもらえますか?

前もって感謝します。

私はExecutor特にフレームワークを使用しています。たとえば100のExecutors.newCachedThreadPool();
リストがあります。 最初の50は、それぞれが最後の50で使用される値（リストに格納されている）を作成 します。それらはリストにあり、同じ順序で実行されます。 しかし、これは起こっていません。 タスクはランダムな順序で実行されているように見え、順番に実行されるのではなく、インターリーブされます。 これはどのように機能するのでしょうか？この問題を回避する方法はありますか？Runnable

Runnableexecutor.execute()

ありがとう

新しい接続を無期限に待機し、それぞれをスレッドで処理する統合サーバーを備えたクライアントを作成しています。

受信したバイト配列を、メイン スレッドのシステム全体で使用可能なメッセージ ハンドラーで処理したいと考えています。ただし、現在、処理は明らかにクライアント スレッドで行われます。

Futures、ExecutorService の submit() を見てきましたが、サーバー内でクライアント接続を作成すると、データがサーバー スレッドに返されます。サーバーをブロックせずに処理するために、そこからメイン スレッド (同期されたパケット ストア内でしょうか?) に戻すにはどうすればよいですか?

私の現在の実装は次のようになります。

更新:堅牢な方法は、TomTom が提案したもののようです - 代わりに新しい java.nio を使用します。このプロジェクトは使用が制限されており、実験的なものであるため、java.io/java.net で使用する最良の方法を知りたいです :)

単一のスレッドが並列で実行できるI/Oを多用するタスクを生成するソリューションをコーディングしようとしています。各タスクには、重要なメモリ内データがあります。そのため、現在保留中のタスクの数を制限できるようにしたいと思います。

このようにThreadPoolExecutorを作成すると、次のようになります。

次に、キューがいっぱいになり、すべてのスレッドがすでにビジー状態になるとexecutor.submit(callable)スローします。RejectedExecutionException

executor.submit(callable)キューがいっぱいですべてのスレッドがビジーのときにブロックを作成するにはどうすればよいですか？

編集：私はこれを試しました：

そして、それは私が達成したい効果をいくらか達成しますが、エレガントではありません（基本的に拒否されたスレッドは呼び出し元のスレッドで実行されるため、呼び出し元のスレッドがそれ以上送信するのをブロックします）。

編集:(質問をしてから5年後）

この質問とその回答を読んでいる人には、受け入れられた回答を1つの正しい解決策と見なさないでください。すべての回答とコメントをお読みください。

問題

ExecutorServiceを介していくつかの外部メソッドの複数の呼び出しを実行しています。これらのメソッドを中断できるようにしたいのですが、残念ながら、それらは割り込みフラグを自分でチェックしません。これらのメソッドから強制的に例外を発生させる方法はありますか?

任意の場所から例外をスローすることは潜在的に危険であることを認識しています。私の特定のケースでは、このチャンスを喜んで利用し、結果に対処する準備ができています。

詳細

「外部メソッド」とは、外部ライブラリに由来するいくつかのメソッドを意味し、そのコードを変更することはできません (できますが、新しいバージョンがリリースされるたびにメンテナンスの悪夢になります)。

外部メソッドは計算コストが高く、IO バウンドではないため、通常の割り込みには応答せず、チャネルやソケットなどを強制的に閉じることはできません。前に述べたように、割り込みフラグもチェックしません。

コードは概念的に次のようなものです。

私が試したこと

Thread.stop()理論的には私が望むことを行いますが、それは非推奨であるだけでなく、エグゼキュータ タスクで作業している間は実際のスレッドでしか利用できません (たとえば、スレッド プールで作業する場合など、将来のタスクとスレッドを共有する可能性もあります)。 . それにもかかわらず、より良い解決策が見つからない場合は、代わりに旧式のスレッドを使用するようにコードを変換し、この方法を使用します。

私が試した別のオプションはmyMethod()、特別な「割り込み可能」アノテーションを使用して同様のメソッドをマークし、AspectJ (私は確かに初心者です) を使用して、そこですべてのメソッド呼び出しをキャッチすることです。

ただしwithincode、一致するメソッドによって呼び出されるメソッドには再帰的ではないため、この注釈を外部コードに編集する必要があります。

最後に、これは私の以前の質問と似ていますが、顕著な違いは、外部ライブラリを扱っていることです。

ScheduledExecutorService を使用して Runnable を定期的に実行するようにスケジュールしたところ、OutOfMemory などのシステム エラーが発生したとします。黙って飲み込んでしまいます。

それは正常ですか？

コンテナに伝播しないのはなぜですか?

このようなエラーを処理する正しい方法は何ですか?

ありがとう！

Runnable を実装するクラスのいくつかのオブジェクトをスケジュールするために、ScheduledThreadPoolExecutor.schedule(Runnable,int,TimeUnit) を使用しています。

ある時点で、アプリケーションがシャットダウンし、ScheduledThreadPoolExecutor.shutdownNow()を使用します。ドキュメントによると、ScheduledFuture のリストを返します。

私が本当にやりたいことは、最初にスケジュールしたオブジェクトを取得し、そこから少しのデータを取得して、実行できなかったことを出力することです。これは、後でアプリケーションがバックアップを開始するときに実行を試みるために使用されます。

ほぼ無限の時間と反復回数で複数のスレッドを実行しています。最良の解決策が見つかった場合、反復回数は0にリセットされます。無限ループを防ぐために、最大反復回数が設定されます。

カウントダウンラッチを使用して、すべてのスレッドが最大反復回数に達したときにプロセスを停止します。つまり、スレッドが最大反復回数に達すると、 notifyThreadStop（）を使用してメインスレッドに通知します。これにより、すべてのスレッドが停止すると、countdown（）がトリガーされます。

注：私のスレッドはFixedThreadPoolExecutorService内で実行されています。

maxTimeラッチを追加したいと思います。だから私がしたことは次のとおりです

これで、カウントダウンによってラッチがトリガーされた場合は、すべてのスレッドが停止しているため、ExecutorServiceをシャットダウンできることがわかりました。

最大時間に達した場合はそうではありません。したがって、（1）では、すべてのランナーを繰り返し処理して、文明的な方法でランナーを終了させたいと思います:-)。そのために、私は関数requestTermination（）を定義しました。これは、簡単に言えば、runnablesでiterationCounterをMaxIterationCountに設定します。

したがって、（1）は

さて、先に進む前に、すべてのスレッドが実際に停止するまでもう一度待つ必要があります。

したがって、（2）は

もちろん、私の関数notifyThreadStop（）はそれを変更する必要があり、mylatchではなくmylatch2でcountdown（）を実行するように指示するフラグが必要になります。

質問に答えたばかりだと思いますが、これがすべて書かれているので、他の人が参照できるようにここに残しておきます。

問題は次のようになります：これを処理するためのより良い方法はありますか？（1）または（2）のshutdownNow（）だけが必要ですか？私のスレッドは、終了する前に、自分のログファイルを閉じて、内側のCallableスレッド*ss*をシャットダウンする必要があることを知っています。

ScheduledExecutorServiceで実行しているときに、タスク自体の内部からタスクの繰り返しを停止するための優れた方法はありますか？

たとえば、次のタスクがあります。

外部からはf.cancel（）で簡単にキャンセルできますが、指定した場所で繰り返しを止めるにはどうすればよいですか？（AtomicReferenceを介してFutureを渡すことは安全ではありません。これは、scheduleAtFixedRateがf lateを返し、変数もlateに設定され、タスク自体がすでに実行されている可能性があり、参照にnullが含まれている可能性があるためです。）