[多線程] 等待所有任務執行完成

在多線程中，有時候需要等待所有線程執行完成后才讓繼續往下執行，如查詢結果匯總等。下面列舉兩種等待方式：

 

方式一：  利用 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() { ... }
 * }}

 

方式二：  利用任務自身的阻塞能力阻塞主線程執行。

示例：

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