使用 Phaser 等待线程池任务完成

Question

只有在提交到线程池的所有任务都完成后，我才想从我的executeTasks() 方法返回。请注意，我的线程池具有可配置的线程池大小并使用SynchronousQueue 作为后备队列，因此我的 for 循环仅在线程可用时通过提交任务来安全地进行。所以，我只想等待最后的任务。我为此使用Phaser。

我已经创建了一个带有 1 个注册方的 Phaser，即当前线程，并且我在向线程池提交任务之前向 Phaser 注册了一个新方，当任务完成时，我取消注册任务方。当 for 循环通过提交最终任务完成时，我希望我的 arriveAndAwaitAdvance() 将等待注册方到达，但它只会在一段时间后发现所有这些方都被取消注册，然后继续前进并从我的方法返回。

我认为这将解决我的问题。如果我错了，或者是否有其他更好的方法，请告诉我。 Countdownlatch 不会有帮助，因为我的线程池大小是可配置的。我知道有一个计数器和监视器可以解决这个问题，但我想要像 Phaser 这样的开箱即用的解决方案。

private void executeTasks(TheadPoolExecutor threadPool, Iterator<String> it) {
    final Phaser phaser = new Phaser(1);

    for (final String id : IteratorUtils.iterable(it)) {
        phaser.register();
        threadPool.execute(() -> {
        // phaser.arrive();
            try {
                thread.sleep(10000 * id.length());
            } finally {
                phaser.arriveAndDeregister();
            }
        });
    }
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndDeregister();
}

Answer 1

我以前从未使用过Phaser，但我认为CountDownLatch 是处理此任务的更好方法。

CountDownLatch 是一个同步屏障，它允许一个或多个线程等待，直到在其他线程中执行的一组操作完成。

在使用CountDownLatch 时有两种方法很有用：

• countDown 在任务完成时递减计数器。
• await 用于当前线程（例如 main）等待其他线程完成。

*

private void executeTasks(TheadPoolExecutor threadPool, Iterator<String> it) {
    final CountDownLatch countDownLatch = new CountDownLatch(threadPool.getPoolSize());

    for (final String id : IteratorUtils.iterable(it)) {
        threadPool.execute(() -> {
            try {
                thread.sleep(10000 * id.length());
                countDownLatch.countDown();
            } catch (InterruptedException ex) {}
        });
    }
    countDownLatch.await();
  }

这里 CountDownLatch 使用线程池中的线程数进行初始化。