[多线程] 等待所有任务执行完成

Posted 2020-12-09 guihong

在多线程中，有时候需要等待所有线程执行完成后才让继续往下执行，如查询结果汇总等。下面列举两种等待方式：

 

方式一：  利用 CountDownLatch 类完成。

示例：

// 按任务总量创建计数器
final CountDownLatch countDownLatch = new CountDownLatch(sourceList.size() - 1);

for (int i = 1; i < sourceList.size(); i++) {
    final int index = i;
    // 线程池提交任务
    threadPoolExecutor.submit(new Runnable() {
        @Override
        public void run() {
            try {// 单个计算
                work();
            } finally {
　　　　　　　　　　// 计数器-1
                countDownLatch.countDown();
            }
        }
    });
}

try {
    // 等待计数器归0
    countDownLatch.await();
} catch (Exception e) {
}

 

JDK源码示例参考：

class Driver { // ...
 *   void main() throws InterruptedException {
 *     CountDownLatch startSignal = new CountDownLatch(1);
 *     CountDownLatch doneSignal = new CountDownLatch(N);
 *
 *     for (int i = 0; i < N; ++i) // create and start threads
 *       new Thread(new Worker(startSignal, doneSignal)).start();
 *
 *     doSomethingElse();            // don‘t let run yet
 *     startSignal.countDown();      // let all threads proceed
 *     doSomethingElse();
 *     doneSignal.await();           // wait for all to finish
 *   }
 * }
 *
 * class Worker implements Runnable {
 *   private final CountDownLatch startSignal;
 *   private final CountDownLatch doneSignal;
 *   Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
 *     this.startSignal = startSignal;
 *     this.doneSignal = doneSignal;
 *   }
 *   public void run() {
 *     try {
 *       startSignal.await();
 *       doWork();
 *       doneSignal.countDown();
 *     } catch (InterruptedException ex) {} // return;
 *   }
 *
 *   void doWork() { ... }
 * }}
 
 * class Driver2 { // ...
 *   void main() throws InterruptedException {
 *     CountDownLatch doneSignal = new CountDownLatch(N);
 *     Executor e = ...
 *
 *     for (int i = 0; i < N; ++i) // create and start threads
 *       e.execute(new WorkerRunnable(doneSignal, i));
 *
 *     doneSignal.await();           // wait for all to finish
 *   }
 * }
 *
 * class WorkerRunnable implements Runnable {
 *   private final CountDownLatch doneSignal;
 *   private final int i;
 *   WorkerRunnable(CountDownLatch doneSignal, int i) {
 *     this.doneSignal = doneSignal;
 *     this.i = i;
 *   }
 *   public void run() {
 *     try {
 *       doWork(i);
 *       doneSignal.countDown();
 *     } catch (InterruptedException ex) {} // return;
 *   }
 *
 *   void doWork() { ... }
 * }}

View Code

 

方式二：  利用任务自身的阻塞能力阻塞主线程执行。

示例：

private ThreadPoolExecutor threadPoolExecutor = null;
private void init() {// 线程池初始化
    BlockingQueue<Runnable> blockingQueue = new LinkedBlockingDeque<Runnable>(30);
    threadPoolExecutor = new ThreadPoolExecutor(1,1, 0, TimeUnit.SECONDS, blockingQueue,
            new ThreadPoolExecutor.CallerRunsPolicy());
}

// 批量任务生成
List<FutureTask<Result<Long>>> taskList = new LinkedList<>();
for (String sql : sqlList) {
    FutureTask<Result<Long>> task = new FutureTask(new Callable<Result<Long>>() {
        @Override
        public Result<Long> call() throws Exception {
            return querySql(sql);
        }
    });
    taskList.add(task);
}

// 批量执行
for (FutureTask<Result<Long>> task : taskList) {
    threadPoolExecutor.execute(task);
}

// 结果汇总
Long cnt = 0L;
for (FutureTask<Result<Long>> task : taskList) {
    try {
        // 任务结果获取，会阻塞主线程等待执行结束
        Result<Long> value = task.get();  
        if (value.isSuccess()) {
            cnt = cnt + value.getValue();
        }
    } catch (Throwable e) {
    }
}

与 submit 类似的，参考：
https://blog.csdn.net/m0_37822338/article/details/100010495