执行程序：如果递归创建任务，如何同步等待所有任务完成？

Question

我的问题与这里的这个问题密切相关。如此处所述，我希望主线程等待，直到工作队列为空并且所有任务都已完成。但是，我的情况是每个任务都可能递归地导致新任务被提交进行处理。这使得收集所有这些任务的未来变得有点尴尬。

我们当前的解决方案使用忙等待循环来等待终止：

        do { //Wait until we are done the processing
      try {
        Thread.sleep(200);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    } while (!executor.getQueue().isEmpty()
             || numTasks.longValue() > executor.getCompletedTaskCount());

numTasks是随着创建每个新任务而增加的值。这可以工作，但是由于繁忙的等待，我认为它不是很好。我想知道是否有一个好方法可以使主线程同步等待，直到被显式唤醒。

Answer 1

非常感谢您的所有建议！

最后，我选择了一些我认为相当简单的东西。我发现CountDownLatch几乎是我所需要的。它一直阻塞直到计数器达到0。唯一的问题是它只能递减计数，不能递减计数，因此不能在我有任务可以提交新任务的动态设置中使用。因此，我实现了一个CountLatch提供附加功能的新类。（请参阅下文）然后，按如下方式使用此类。

主线程调用latch.awaitZero()，直到锁存器到达0为止。

任何线程，在调用executor.execute(..)calls 之前latch.increment()。

在完成之前，任何任务都会调用latch.decrement()。

当最后一个任务终止时，计数器将达到0，从而释放主线程。

欢迎进一步的建议和反馈！

public class CountLatch {

@SuppressWarnings("serial")
private static final class Sync extends AbstractQueuedSynchronizer {

    Sync(int count) {
        setState(count);
    }

    int getCount() {
        return getState();
    }

    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

    protected int acquireNonBlocking(int acquires) {
        // increment count
        for (;;) {
            int c = getState();
            int nextc = c + 1;
            if (compareAndSetState(c, nextc))
                return 1;
        }
    }

    protected boolean tryReleaseShared(int releases) {
        // Decrement count; signal when transition to zero
        for (;;) {
            int c = getState();
            if (c == 0)
                return false;
            int nextc = c - 1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
}

private final Sync sync;

public CountLatch(int count) {
    this.sync = new Sync(count);
}

public void awaitZero() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

public boolean awaitZero(long timeout, TimeUnit unit) throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

public void increment() {
    sync.acquireNonBlocking(1);
}

public void decrement() {
    sync.releaseShared(1);
}

public String toString() {
    return super.toString() + "[Count = " + sync.getCount() + "]";
}

}

请注意，可以将increment()/ decrement()调用封装到自定义的Executor子类中，如Sami Korhonen
所建议的那样，或者与beforeExecute和afterExecute一起由impl所建议的那样。看这里：

public class CountingThreadPoolExecutor extends ThreadPoolExecutor {

protected final CountLatch numRunningTasks = new CountLatch(0);

public CountingThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
        BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}

@Override
public void execute(Runnable command) {
    numRunningTasks.increment();
    super.execute(command);
}

@Override
protected void afterExecute(Runnable r, Throwable t) {
    numRunningTasks.decrement();
    super.afterExecute(r, t);
}

/**
 * Awaits the completion of all spawned tasks.
 */
public void awaitCompletion() throws InterruptedException {
    numRunningTasks.awaitZero();
}

/**
 * Awaits the completion of all spawned tasks.
 */
public void awaitCompletion(long timeout, TimeUnit unit) throws InterruptedException {
    numRunningTasks.awaitZero(timeout, unit);
}

}