我使用ExecutorService 来执行任务。这个任务可以递归地创建提交给同一个ExecutorService的其他任务，这些子任务也可以这样做。
我现在有一个问题，我想等到所有任务都完成（即所有任务都完成并且他们没有提交新任务）后再继续。
我不能在主线程中调用ExecutorService.shutdown()，因为这会阻止ExecutorService 接受新任务。
如果 shutdown 没有被调用，调用 ExecutorService.awaitTermination() 似乎没有任何作用。
所以我有点卡在这里。 ExecutorService 看到所有工作人员都处于空闲状态并不是那么难，不是吗？我能想出的唯一不优雅的解决方案是直接使用ThreadPoolExecutor 并每隔一段时间查询它的getPoolSize()。真的没有更好的办法吗？
【问题讨论】：
标签： java multithreading concurrency executorservice
您可以使用跟踪正在运行的线程的运行器：
Runner runner = Runner.runner(numberOfThreads);
runner.runIn(2, SECONDS, callable);
runner.run(callable);
// blocks until all tasks are finished (or failed)
runner.waitTillDone();
// and reuse it
runner.runRunnableIn(500, MILLISECONDS, runnable);
runner.waitTillDone();
// and then just kill it
runner.shutdownAndAwaitTermination();
要使用它，您只需添加一个依赖项：
编译'com.github.matejtymes:javafixes:1.3.0'
【讨论】：
我能想出的唯一不优雅的解决方案是直接使用 ThreadPoolExecutor 并每隔一段时间查询它的 getPoolSize()。真的没有更好的办法吗？
您必须以正确的顺序使用shutdown() ,awaitTermination()and shutdownNow() 方法。
shutdown()：启动有序关闭，执行之前提交的任务，但不会接受新任务。
awaitTermination()：在关闭请求后阻塞，直到所有任务都完成执行，或者发生超时，或者当前线程被中断，以先发生者为准。
shutdownNow()：尝试停止所有正在执行的任务，停止等待任务的处理，并返回等待执行的任务列表。
来自ExecutorService的oracle文档页面的推荐方式：
void shutdownAndAwaitTermination(ExecutorService pool) {
pool.shutdown(); // Disable new tasks from being submitted
try {
// Wait a while for existing tasks to terminate
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
pool.shutdownNow(); // Cancel currently executing tasks
// Wait a while for tasks to respond to being cancelled
if (!pool.awaitTermination(60, TimeUnit.SECONDS))
System.err.println("Pool did not terminate");
}
} catch (InterruptedException ie) {
// (Re-)Cancel if current thread also interrupted
pool.shutdownNow();
// Preserve interrupt status
Thread.currentThread().interrupt();
}
如果完成以下任务的持续时间较长，您可以将if条件替换为while条件：
改变
if (!pool.awaitTermination(60, TimeUnit.SECONDS))
到
while(!pool.awaitTermination(60, TimeUnit.SECONDS)) {
Thread.sleep(60000);
}
您可以参考其他替代方案（join() 除外，可与独立线程一起使用）：
【讨论】：
您可以创建自己的线程池来扩展ThreadPoolExecutor。您想知道任务何时提交以及何时完成。
public class MyThreadPoolExecutor extends ThreadPoolExecutor {
private int counter = 0;
public MyThreadPoolExecutor() {
super(1, 1, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
}
@Override
public synchronized void execute(Runnable command) {
counter++;
super.execute(command);
}
@Override
protected synchronized void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
counter--;
notifyAll();
}
public synchronized void waitForExecuted() throws InterruptedException {
while (counter == 0)
wait();
}
}
【讨论】：
为您的任务使用Future（而不是提交Runnable），回调会在完成时更新其状态，因此您可以使用Future.isDone 跟踪所有任务的状态。
【讨论】：
使用CountDownLatch。 将 CountDownLatch 对象传递给您的每个任务，并为您的任务编写如下代码。
public void doTask() {
// do your task
latch.countDown();
}
而需要等待的线程应该执行以下代码：
public void doWait() {
latch.await();
}
当然，这假设您已经知道子任务的数量，以便您可以初始化锁存器的计数。
【讨论】：
我必须说，上面描述的递归调用任务和等待结束子任务问题的解决方案并不能让我满意。我的解决方案灵感来自 Oracle 的原始文档：CountDownLatch 和示例：Human resources CountDownLatch。
类 HRManagerCompact 实例中的第一个公共线程有两个子线程的等待锁存器，其后续 2 个子线程的等待锁存器......等等。
当然，latch可以设置为不等于2的值（在CountDownLatch的构造函数中），也可以在迭代中建立可运行对象的数量即ArrayList，但必须对应（倒计时的数量必须是等于 CountDownLatch 构造函数中的参数）。
注意，锁存器的数量会根据限制条件呈指数增长： 'level.get()
package processes.countdownlatch.hr;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
/** Recursively latching running classes to wait for the peak threads
*
* @author hariprasad
*/
public class HRManagerCompact extends Thread {
final int N = 2; // number of daughter's tasks for latch
CountDownLatch countDownLatch;
CountDownLatch originCountDownLatch;
AtomicInteger level = new AtomicInteger(0);
AtomicLong order = new AtomicLong(0); // id latched thread waiting for
HRManagerCompact techLead1 = null;
HRManagerCompact techLead2 = null;
HRManagerCompact techLead3 = null;
// constructor
public HRManagerCompact(CountDownLatch countDownLatch, String name,
AtomicInteger level, AtomicLong order){
super(name);
this.originCountDownLatch=countDownLatch;
this.level = level;
this.order = order;
}
private void doIt() {
countDownLatch = new CountDownLatch(N);
AtomicInteger leveli = new AtomicInteger(level.get() + 1);
AtomicLong orderi = new AtomicLong(Thread.currentThread().getId());
techLead1 = new HRManagerCompact(countDownLatch, "first", leveli, orderi);
techLead2 = new HRManagerCompact(countDownLatch, "second", leveli, orderi);
//techLead3 = new HRManagerCompact(countDownLatch, "third", leveli);
techLead1.start();
techLead2.start();
//techLead3.start();
try {
synchronized (Thread.currentThread()) { // to prevent print and latch in the same thread
System.out.println("*** HR Manager waiting for recruitment to complete... " + level + ", " + order + ", " + orderi);
countDownLatch.await(); // wait actual thread
}
System.out.println("*** Distribute Offer Letter, it means finished. " + level + ", " + order + ", " + orderi);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void run() {
try {
System.out.println(Thread.currentThread().getName() + ": working... " + level + ", " + order + ", " + Thread.currentThread().getId());
Thread.sleep(10*level.intValue());
if (level.get() < 2) doIt();
Thread.yield();
}
catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
/*catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}*/
// TODO Auto-generated method stub
System.out.println("--- " +Thread.currentThread().getName() + ": recruted " + level + ", " + order + ", " + Thread.currentThread().getId());
originCountDownLatch.countDown(); // count down
}
public static void main(String args[]){
AtomicInteger levelzero = new AtomicInteger(0);
HRManagerCompact hr = new HRManagerCompact(null, "zero", levelzero, new AtomicLong(levelzero.longValue()));
hr.doIt();
}
}
可能的注释输出（有一定的概率）：
first: working... 1, 1, 10 // thread 1, first daughter's task (10)
second: working... 1, 1, 11 // thread 1, second daughter's task (11)
first: working... 2, 10, 12 // thread 10, first daughter's task (12)
first: working... 2, 11, 14 // thread 11, first daughter's task (14)
second: working... 2, 11, 15 // thread 11, second daughter's task (15)
second: working... 2, 10, 13 // thread 10, second daughter's task (13)
--- first: recruted 2, 10, 12 // finished 12
--- first: recruted 2, 11, 14 // finished 14
--- second: recruted 2, 10, 13 // finished 13 (now can be opened latch 10)
--- second: recruted 2, 11, 15 // finished 15 (now can be opened latch 11)
*** HR Manager waiting for recruitment to complete... 0, 0, 1
*** HR Manager waiting for recruitment to complete... 1, 1, 10
*** Distribute Offer Letter, it means finished. 1, 1, 10 // latch on 10 opened
--- first: recruted 1, 1, 10 // finished 10
*** HR Manager waiting for recruitment to complete... 1, 1, 11
*** Distribute Offer Letter, it means finished. 1, 1, 11 // latch on 11 opened
--- second: recruted 1, 1, 11 // finished 11 (now can be opened latch 1)
*** Distribute Offer Letter, it means finished. 0, 0, 1 // latch on 1 opened
【讨论】：
这确实是 Phaser 的理想人选。 Java 7 即将推出这个新类。它是一个灵活的 CountdonwLatch/CyclicBarrier。您可以在JSR 166 Interest Site 获得稳定版本。
它是更灵活的 CountdownLatch/CyclicBarrier 的方式是因为它不仅能够支持未知数量的参与方（线程），而且还可以重复使用（这就是阶段部分的来源）
对于您提交的每项任务，您都会注册，当该任务完成时，您就会到达。这可以递归完成。
Phaser phaser = new Phaser();
ExecutorService e = //
Runnable recursiveRunnable = new Runnable(){
public void run(){
//do work recursively if you have to
if(shouldBeRecursive){
phaser.register();
e.submit(recursiveRunnable);
}
phaser.arrive();
}
}
public void doWork(){
int phase = phaser.getPhase();
phaser.register();
e.submit(recursiveRunnable);
phaser.awaitAdvance(phase);
}
编辑：感谢@depthofreality 在我之前的示例中指出竞争条件。我正在更新它，以便执行线程仅等待当前阶段的推进，因为它阻塞递归函数完成。
阶段号直到arrives == registers 的数量才会跳闸。由于在每次递归调用调用register 之前，当所有调用完成时都会发生阶段增量。
【讨论】：
doWork 中完成，因为下面的移相器不需要arriveAndAwaitAdvance recursiveRunnable 需要到达（确实如此）。在执行到 ExecutorService 之前还要注意可运行的registers
如果您想使用 JSR166y 类 - 例如Phaser 或 Fork/Join——其中任何一个都可能适合您，您始终可以从以下位置下载它们的 Java 6 反向移植：http://gee.cs.oswego.edu/dl/concurrency-interest/ 并将其用作基础，而不是编写完全自制的解决方案。然后当 7 出现时，您可以删除对 backport 的依赖并更改一些包名称。
（完全披露：我们已经在 prod 中使用 LinkedTransferQueue 有一段时间了。没有问题）
【讨论】：
（mea culpa：我的就寝时间有点过了；）但这是第一次尝试动态闩锁）：
package oss.alphazero.sto4958330;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
public class DynamicCountDownLatch {
@SuppressWarnings("serial")
private static final class Sync extends AbstractQueuedSynchronizer {
private final CountDownLatch toplatch;
public Sync() {
setState(0);
this.toplatch = new CountDownLatch(1);
}
@Override
protected int tryAcquireShared(int acquires){
try {
toplatch.await();
}
catch (InterruptedException e) {
throw new RuntimeException("Interrupted", e);
}
return getState() == 0 ? 1 : -1;
}
public boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
public boolean tryExtendState(int acquires) {
for (;;) {
int s = getState();
int exts = s+1;
if (compareAndSetState(s, exts)) {
toplatch.countDown();
return exts > 0;
}
}
}
}
private final Sync sync;
public DynamicCountDownLatch(){
this.sync = new Sync();
}
public void await()
throws InterruptedException
{
sync.acquireSharedInterruptibly(1);
}
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException
{
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
public void countDown() {
sync.releaseShared(1);
}
public void join() {
sync.tryExtendState(1);
}
}
此锁存器为现有（克隆的）CountDownLatch API 引入了一个新方法 join()，任务使用该方法表示它们进入更大的任务组。
latch 从父任务传递到子任务。根据 Suraj 的模式，每个任务都会首先“join()”锁存器，执行其 task()，然后 countDown()。
为了解决主线程启动任务组然后立即等待（）的情况——在任何任务线程甚至有机会加入（）之前——topLatch 用于 int 内部 Sync班级。这是一个锁存器，每次 join() 都会被倒计时；当然，只有第一个倒计时很重要，因为所有后续倒计时都是 nops。
上面的初始实现确实引入了某种语义上的皱纹，因为 tryAcquiredShared(int) 不应该抛出 InterruptedException，但我们确实需要处理 topLatch 上的等待中断。
这是对 OP 自己使用原子计数器的解决方案的改进吗？我会说可能不是 IFF，他坚持使用 Executors，但我相信，在这种情况下，它是使用 AQS 的同样有效的替代方法，并且也可用于通用线程。
击退其他黑客。
【讨论】：
如果递归任务树中的任务数量最初是未知的，也许最简单的方法是实现您自己的同步原语，某种“反向信号量”，并在您的任务之间共享它。在提交每个任务之前，您增加一个值，当任务完成时，它会减少该值，然后等待该值变为 0。
将其实现为从任务中显式调用的单独原语可以将此逻辑与线程池实现分离，并允许您将多个独立的递归任务树提交到同一个池中。
类似这样的：
public class InverseSemaphore {
private int value = 0;
private Object lock = new Object();
public void beforeSubmit() {
synchronized(lock) {
value++;
}
}
public void taskCompleted() {
synchronized(lock) {
value--;
if (value == 0) lock.notifyAll();
}
}
public void awaitCompletion() throws InterruptedException {
synchronized(lock) {
while (value > 0) lock.wait();
}
}
}
请注意，taskCompleted() 应在 finally 块内调用，以使其免受可能的异常的影响。
还要注意beforeSubmit()应该在提交任务之前由提交线程调用，而不是由任务本身调用，以避免在旧任务完成而新任务尚未启动时可能出现的“错误完成”。
编辑：修复了使用模式的重要问题。
【讨论】：
哇，你们真快:)
感谢您的所有建议。 Futures 不容易与我的模型集成，因为我不知道事先安排了多少可运行对象。因此，如果我让父任务保持活动状态只是为了等待它的递归子任务完成，那么我周围就会有很多垃圾。
我使用 AtomicInteger 建议解决了我的问题。本质上，我将 ThreadPoolExecutor 子类化，并在调用 execute() 时递增计数器，并在调用 afterExecute() 时递减计数器。当计数器变为 0 时，我调用 shutdown()。这似乎对我的问题有用，不确定这是否是一种普遍的好方法。特别是，我假设您只使用 execute() 来添加 Runnables。
作为一个侧节点：我首先尝试在 afterExecute() 中检查队列中的 Runnable 数量以及当它们为 0 时处于活动状态和关闭的工作人员数量；但这不起作用，因为并非所有 Runnable 都出现在队列中，而且 getActiveCount() 也没有达到我的预期。
无论如何，这是我的解决方案：（如果有人发现此问题严重，请告诉我：）
public class MyThreadPoolExecutor extends ThreadPoolExecutor {
private final AtomicInteger executing = new AtomicInteger(0);
public MyThreadPoolExecutor(int coorPoolSize, int maxPoolSize, long keepAliveTime,
TimeUnit seconds, BlockingQueue<Runnable> queue) {
super(coorPoolSize, maxPoolSize, keepAliveTime, seconds, queue);
}
@Override
public void execute(Runnable command) {
//intercepting beforeExecute is too late!
//execute() is called in the parent thread before it terminates
executing.incrementAndGet();
super.execute(command);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
int count = executing.decrementAndGet();
if(count == 0) {
this.shutdown();
}
}
}
【讨论】：