java - ExecutorService、すべてのタスクが完了するのを待つ方法

Question

のすべてのタスクが完了するのを待つ最も簡単な方法は何ExecutorServiceですか? 私のタスクは主に計算なので、各コアで 1 つずつ、多数のジョブを実行したいだけです。現在、私のセットアップは次のようになっています。

ExecutorService es = Executors.newFixedThreadPool(2);
for (DataTable singleTable : uniquePhrases) {   
    es.execute(new ComputeDTask(singleTable));
}
try{
    es.wait();
} 
catch (InterruptedException e){
    e.printStackTrace();
}

ComputeDTask実行可能を実装します。これにより、タスクが正しく実行されているように見えますが、コードはでクラッシュwait()しIllegalMonitorStateExceptionます。おもちゃの例をいくつか試してみたところ、うまくいくように見えたので、これは奇妙です。

uniquePhrases数万の要素が含まれています。別の方法を使用する必要がありますか? できるだけシンプルなものを探しています

score 240 · Accepted Answer

最も簡単なアプローチはExecutorService.invokeAll()、ワンライナーで必要なことをする which を使用することです。あなたの言い方では、ComputeDTask実装するために変更またはラップする必要がありCallable<>ます。これにより、柔軟性が大幅に向上します。おそらくあなたのアプリにはの意味のある実装がありますが、Callable.call()を使用しない場合はラップする方法がありますExecutors.callable()。

ExecutorService es = Executors.newFixedThreadPool(2);
List<Callable<Object>> todo = new ArrayList<Callable<Object>>(singleTable.size());

for (DataTable singleTable: uniquePhrases) { 
    todo.add(Executors.callable(new ComputeDTask(singleTable))); 
}

List<Future<Object>> answers = es.invokeAll(todo);

他の人が指摘したように、invokeAll()適切な場合はタイムアウトバージョンを使用できます。この例では、 null を返す一連のがanswers含まれます( の定義を参照してください。おそらく、やりたいことはわずかなリファクタリングであり、有用な回答を取得したり、基になるへの参照を取得したりできますが、あなたの例からはわかりません。FutureExecutors.callable()ComputeDTask

不明な場合は、invokeAll()すべてのタスクが完了するまで戻らないことに注意してください。(つまり、コレクションFuture内のすべての s は、要求された場合answersに報告します.isDone()。) これにより、すべての手動シャットダウン、awaitTermination などが回避されExecutorService、必要に応じて複数のサイクルでこれをきちんと再利用できます。

SO に関するいくつかの関連する質問があります。

これらはどれもあなたの質問に厳密に当てはまるものではありませんが、人々がどのように考えているか、どのように使用すべきかについて少し色を付けていExecutorますExecutorService.

score 66 · Accepted Answer

すべてのタスクが完了するまで待機する場合は、のshutdown代わりにメソッドを使用しwaitます。その後、でフォローしawaitTerminationます。

また、Runtime.availableProcessorsスレッドプールを適切に初期化できるように、ハードウェアスレッドの数を取得するために使用できます。

score 49 · Accepted Answer

のすべてのタスクがExecutorService完了するのを待つことが正確な目標ではなく、タスクの特定のバッチが完了するまで待つ場合は、を使用できます。CompletionService具体的にはExecutorCompletionService.

アイデアは、ExecutorCompletionServiceラップを作成し、を介して既知の数のタスクをExecutor送信し、 (ブロックする) または(ブロックしない)のいずれかを使用して完了キューから同じ数の結果を引き出すことです。提出したタスクに対応するすべての期待される結果を描画したら、それらがすべて完了したことがわかります。CompletionServicetake()poll()

CompletionServiceインターフェースからは明らかではないので、もう一度述べさせてください:引き出そうとするものの数を知るためには、に入れるものの数を知る必要があります。これは特にメソッドで重要take()です。一度呼び出すと、他のスレッドが別のジョブを同じに送信するまで、呼び出し元のスレッドがブロックされますCompletionService。

本Java Concurrency in Practiceには、使用方法を示すいくつかの例CompletionServiceがあります。

score 13 · Accepted Answer

executor サービスの実行が完了するのを待ちたい場合は、呼び出しshutdown()てからawaitTermination(units , unitType)awaitTermination(1, MINUTE)を呼び出します。ExecutorService は独自のモニターでブロックしないため、使用できませんwait。

score 8 · Accepted Answer

特定の間隔でジョブが完了するのを待つことができます。

int maxSecondsPerComputeDTask = 20;
try {
    while (!es.awaitTermination(uniquePhrases.size() * maxSecondsPerComputeDTask, TimeUnit.SECONDS)) {
        // consider giving up with a 'break' statement under certain conditions
    }
} catch (InterruptedException e) {
    throw new RuntimeException(e);    
}

または、 ExecutorServiceを使用できます。submit ( Runnable ) を実行し、返されたFutureオブジェクトを収集し、それぞれに対して順番にget()を呼び出して、それらが終了するのを待ちます。

ExecutorService es = Executors.newFixedThreadPool(2);
Collection<Future<?>> futures = new LinkedList<<Future<?>>();
for (DataTable singleTable : uniquePhrases) {
    futures.add(es.submit(new ComputeDTask(singleTable)));
}
for (Future<?> future : futures) {
   try {
       future.get();
   } catch (InterruptedException e) {
       throw new RuntimeException(e);
   } catch (ExecutionException e) {
       throw new RuntimeException(e);
   }
}

InterruptedExceptionは、適切に処理するために非常に重要です。これは、あなたやライブラリのユーザーが長いプロセスを安全に終了できるようにするものです。

score 7 · Accepted Answer

使うだけ

latch = new CountDownLatch(noThreads)

各スレッドで

latch.countDown();

そして障壁として

latch.await();

score 6 · Accepted Answer

メソッドを使用できますExecutorService.invokeAll。すべてのタスクを実行し、すべてのスレッドがタスクを完了するまで待機します。

ここに完全なjavadocがあります

このメソッドのオーバーロードバージョンを使用して、タイムアウトを指定することもできます。

ここにサンプルコードがありますExecutorService.invokeAll

public class Test {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService service = Executors.newFixedThreadPool(3);
        List<Callable<String>> taskList = new ArrayList<>();
        taskList.add(new Task1());
        taskList.add(new Task2());
        List<Future<String>> results = service.invokeAll(taskList);
        for (Future<String> f : results) {
            System.out.println(f.get());
        }
    }

}

class Task1 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(2000);
            return "Task 1 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task1";
        }
    }
}

class Task2 implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            Thread.sleep(3000);
            return "Task 2 done";
        } catch (Exception e) {
            e.printStackTrace();
            return " error in task2";
        }
    }
}

score 3 · Accepted Answer

クロールする一連のドキュメントがあるという状況もあります。処理する必要がある最初の「シード」ドキュメントから始めます。そのドキュメントには、処理する必要がある他のドキュメントへのリンクが含まれています。

Crawler私のメインプログラムでは、一連のスレッドを制御する次のようなものを書きたいだけです。

Crawler c = new Crawler();
c.schedule(seedDocument); 
c.waitUntilCompletion()

ツリーをナビゲートしたい場合も同じ状況が発生します。私はルートノードをポップし、各ノードのプロセッサは必要に応じて子をキューに追加し、スレッドの束がツリー内のすべてのノードを処理し、ノードがなくなるまで処理します。

JVM には何も見つかりませんでしたが、これは少し驚くべきことでした。そこでThreadPool、ドメインに適したメソッドを追加するために直接またはサブクラスを使用できるクラスを作成しましたschedule(Document)。それが役に立てば幸い！

スレッドプール Javadoc | メイヴン

score 2 · Accepted Answer

コレクションにすべてのスレッドを追加し、を使用して送信しinvokeAllます。invokeAllのメソッドを使用できる場合ExecutorService、JVM はすべてのスレッドが完了するまで次の行に進みません。

ここに良い例があります: ExecutorService 経由の invokeAll

score 1 · Accepted Answer

タスクをランナーに送信してから、次のようにメソッドwaitTillDone()を呼び出すのを待ちます。

Runner runner = Runner.runner(2);

for (DataTable singleTable : uniquePhrases) {

    runner.run(new ComputeDTask(singleTable));
}

// blocks until all tasks are finished (or failed)
runner.waitTillDone();

runner.shutdown();

それを使用するには、この gradle/maven 依存関係を追加します。'com.github.matejtymes:javafixes:1.0'

詳細については、こちらをご覧ください: https://github.com/MatejTymes/JavaFixesまたはこちら: http://matejtymes.blogspot.com/2016/04/executor-that-notifies-you-when-task.html

score 0 · Accepted Answer

ForkJoinPoolタスクを実行するには、グローバルプールが必要であり、それを使用しているようです。

public static void main(String[] args) {
    // the default `commonPool` should be sufficient for many cases.
    ForkJoinPool pool = ForkJoinPool.commonPool(); 
    // The root of your task that may spawn other tasks. 
    // Make sure it submits the additional tasks to the same executor that it is in.
    Runnable rootTask = new YourTask(pool); 
    pool.execute(rootTask);
    pool.awaitQuiescence(...);
    // that's it.
}

美しさはpool.awaitQuiescence、メソッドが呼び出し元のスレッドをブロックしてそのタスクを実行し、実際に空になったときに戻るところにあります。

score 0 · Accepted Answer

タスクの完了に適していると思われる指定されたタイムアウトでエグゼキュータが終了するのを待ちます。

 try {  
         //do stuff here 
         exe.execute(thread);
    } finally {
        exe.shutdown();
    }
    boolean result = exe.awaitTermination(4, TimeUnit.HOURS);
    if (!result)

    {
        LOGGER.error("It took more than 4 hour for the executor to stop, this shouldn't be the normal behaviour.");
    }

score -1 · Accepted Answer

これに代わる簡単な方法は、結合とともにスレッドを使用することです。参照 :スレッドの結合

java - ExecutorService、すべてのタスクが完了するのを待つ方法

16 に答える 16

Related

Reference