如何判斷線程池任務(wù)已執(zhí)行完？
無論是在項(xiàng)目開發(fā)中，還是在面試中過程中，總會(huì)被問到或使用到并發(fā)編程來完成項(xiàng)目中的某個(gè)功能。
例如某個(gè)復(fù)雜的查詢，無法使用一個(gè)查詢語句來完成此功能，此時(shí)我們就需要執(zhí)行多個(gè)查詢語句，然后再將各自查詢的結(jié)果，組裝之后返回給前端了，那么這種場(chǎng)景下，我們就必須使用線程池來進(jìn)行并發(fā)查詢了。
PS：磊哥做的最復(fù)雜的查詢，總共關(guān)聯(lián)了 21 張表，在和產(chǎn)品及需求方的溝通多次溝通下，才將查詢的業(yè)務(wù)從 21 張表，降到了至少要查詢 12 張表（非常難搞），那么這種場(chǎng)景下是無法使用一個(gè)查詢語句來實(shí)現(xiàn)的，那么并發(fā)查詢是必須要給安排上的。
1、需求分析
線程池的使用并不復(fù)雜，麻煩的是如何判斷線程池中的任務(wù)已經(jīng)全部執(zhí)行完了？因?yàn)槲覀円人腥蝿?wù)都執(zhí)行完之后，才能進(jìn)行數(shù)據(jù)的組裝和返回，所以接下來，我們就來看如何判斷線程中的任務(wù)是否已經(jīng)全部執(zhí)行完？
2、實(shí)現(xiàn)概述
判斷線程池中的任務(wù)是否執(zhí)行完的方法有很多，比如以下幾個(gè)：
- 使用 getCompletedTaskCount() 統(tǒng)計(jì)已經(jīng)執(zhí)行完的任務(wù)，和 getTaskCount() 線程池的總?cè)蝿?wù)進(jìn)行對(duì)比，如果相等則說明線程池的任務(wù)執(zhí)行完了，否則既未執(zhí)行完。
- 使用 FutureTask 等待所有任務(wù)執(zhí)行完，線程池的任務(wù)就執(zhí)行完了。
- 使用 CountDownLatch 或 CyclicBarrier 等待所有線程都執(zhí)行完之后，再執(zhí)行后續(xù)流程。
具體實(shí)現(xiàn)代碼如下。
3、具體實(shí)現(xiàn)
（1）統(tǒng)計(jì)完成任務(wù)數(shù)
通過判斷線程池中的計(jì)劃執(zhí)行任務(wù)數(shù)和已完成任務(wù)數(shù)，來判斷線程池是否已經(jīng)全部執(zhí)行完，如果計(jì)劃執(zhí)行任務(wù)數(shù)=已完成任務(wù)數(shù)，那么線程池的任務(wù)就全部執(zhí)行完了，否則就未執(zhí)行完。示例代碼如下：
private static void isCompletedByTaskCount(ThreadPoolExecutor threadPool) {
while (threadPool.getTaskCount() != threadPool.getCompletedTaskCount()) {
}
}以上程序執(zhí)行結(jié)果如下：
方法說明
- getTaskCount()：返回計(jì)劃執(zhí)行的任務(wù)總數(shù)。由于任務(wù)和線程的狀態(tài)可能在計(jì)算過程中動(dòng)態(tài)變化，因此返回的值只是一個(gè)近似值。
- getCompletedTaskCount()：返回完成執(zhí)行任務(wù)的總數(shù)。因?yàn)槿蝿?wù)和線程的狀態(tài)可能在計(jì)算過程中動(dòng)態(tài)地改變，所以返回的值只是一個(gè)近似值，但是在連續(xù)的調(diào)用中并不會(huì)減少。
缺點(diǎn)分析
此判斷方法的缺點(diǎn)是 getTaskCount() 和 getCompletedTaskCount() 返回的是一個(gè)近似值，因?yàn)榫€程池中的任務(wù)和線程的狀態(tài)可能在計(jì)算過程中動(dòng)態(tài)變化，所以它們兩個(gè)返回的都是一個(gè)近似值。
（2）FutureTask
FutrueTask 的優(yōu)勢(shì)是任務(wù)判斷精準(zhǔn)，調(diào)用每個(gè) FutrueTask 的 get 方法就是等待該任務(wù)執(zhí)行完，如下代碼所示：
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
/**
* 使用 FutrueTask 等待線程池執(zhí)行完全部任務(wù)
*/
public class FutureTaskDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 創(chuàng)建一個(gè)固定大小的線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 創(chuàng)建任務(wù)
FutureTask<Integer> task1 = new FutureTask<>(() -> {
System.out.println("Task 1 start");
Thread.sleep(2000);
System.out.println("Task 1 end");
return 1;
});
FutureTask<Integer> task2 = new FutureTask<>(() -> {
System.out.println("Task 2 start");
Thread.sleep(3000);
System.out.println("Task 2 end");
return 2;
});
FutureTask<Integer> task3 = new FutureTask<>(() -> {
System.out.println("Task 3 start");
Thread.sleep(1500);
System.out.println("Task 3 end");
return 3;
});
// 提交三個(gè)任務(wù)給線程池
executor.submit(task1);
executor.submit(task2);
executor.submit(task3);
// 等待所有任務(wù)執(zhí)行完畢并獲取結(jié)果
int result1 = task1.get();
int result2 = task2.get();
int result3 = task3.get();
System.out.println("Do main thread.");
}
}以上程序的執(zhí)行結(jié)果如下：
（3）CountDownLatch和CyclicBarrier
CountDownLatch 和 CyclicBarrier 類似，都是等待所有任務(wù)到達(dá)某個(gè)點(diǎn)之后，再進(jìn)行后續(xù)的操作，如下圖所示：
CountDownLatch 使用的示例代碼如下：
public static void main(String[] args) throws InterruptedException {
// 創(chuàng)建線程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20,
0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024));
final int taskCount = 5; // 任務(wù)總數(shù)
// 單次計(jì)數(shù)器
CountDownLatch countDownLatch = new CountDownLatch(taskCount); // ①
// 添加任務(wù)
for (int i = 0; i < taskCount; i++) {
final int finalI = i;
threadPool.submit(new Runnable() {
@Override
public void run() {
try {
// 隨機(jī)休眠 0-4s
int sleepTime = new Random().nextInt(5);
TimeUnit.SECONDS.sleep(sleepTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(String.format("任務(wù)%d執(zhí)行完成", finalI));
// 線程執(zhí)行完，計(jì)數(shù)器 -1
countDownLatch.countDown(); // ②
}
});
}
// 阻塞等待線程池任務(wù)執(zhí)行完
countDownLatch.await(); // ③
// 線程池執(zhí)行完
System.out.println();
System.out.println("線程池任務(wù)執(zhí)行完成！");
}代碼說明：以上代碼中標(biāo)識(shí)為 ①、②、③ 的代碼行是核心實(shí)現(xiàn)代碼，其中：① 是聲明一個(gè)包含了 5 個(gè)任務(wù)的計(jì)數(shù)器；② 是每個(gè)任務(wù)執(zhí)行完之后計(jì)數(shù)器 -1；③ 是阻塞等待計(jì)數(shù)器 CountDownLatch 減為 0，表示任務(wù)都執(zhí)行完了，可以執(zhí)行 await 方法后面的業(yè)務(wù)代碼了。
以上程序的執(zhí)行結(jié)果如下：
缺點(diǎn)分析
CountDownLatch 缺點(diǎn)是計(jì)數(shù)器只能使用一次，CountDownLatch 創(chuàng)建之后不能被重復(fù)使用。CyclicBarrier 和 CountDownLatch 類似，它可以理解為一個(gè)可以重復(fù)使用的循環(huán)計(jì)數(shù)器，CyclicBarrier 可以調(diào)用 reset 方法將自己重置到初始狀態(tài)，CyclicBarrier 具體實(shí)現(xiàn)代碼如下：
public static void main(String[] args) throws InterruptedException {
// 創(chuàng)建線程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20,
0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024));
final int taskCount = 5; // 任務(wù)總數(shù)
// 循環(huán)計(jì)數(shù)器 ①
CyclicBarrier cyclicBarrier = new CyclicBarrier(taskCount, new Runnable() {
@Override
public void run() {
// 線程池執(zhí)行完
System.out.println();
System.out.println("線程池所有任務(wù)已執(zhí)行完！");
}
});
// 添加任務(wù)
for (int i = 0; i < taskCount; i++) {
final int finalI = i;
threadPool.submit(new Runnable() {
@Override
public void run() {
try {
// 隨機(jī)休眠 0-4s
int sleepTime = new Random().nextInt(5);
TimeUnit.SECONDS.sleep(sleepTime);
System.out.println(String.format("任務(wù)%d執(zhí)行完成", finalI));
// 線程執(zhí)行完
cyclicBarrier.await(); // ②
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
});
}
}以上程序的執(zhí)行結(jié)果如下：
方法說明
CyclicBarrier 有 3 個(gè)重要的方法：
- 構(gòu)造方法：構(gòu)造方法可以傳遞兩個(gè)參數(shù)，參數(shù) 1 是計(jì)數(shù)器的數(shù)量 parties，參數(shù) 2 是計(jì)數(shù)器為 0 時(shí)，也就是任務(wù)都執(zhí)行完之后可以執(zhí)行的事件（方法）。
- await 方法：在 CyclicBarrier 上進(jìn)行阻塞等待，當(dāng)調(diào)用此方法時(shí) CyclicBarrier 的內(nèi)部計(jì)數(shù)器會(huì) -1，直到發(fā)生以下情形之一：
- 在 CyclicBarrier 上等待的線程數(shù)量達(dá)到 parties，也就是計(jì)數(shù)器的聲明數(shù)量時(shí)，則所有線程被釋放，繼續(xù)執(zhí)行。
- 當(dāng)前線程被中斷，則拋出 InterruptedException 異常，并停止等待，繼續(xù)執(zhí)行。
- 其他等待的線程被中斷，則當(dāng)前線程拋出 BrokenBarrierException 異常，并停止等待，繼續(xù)執(zhí)行。
- 其他等待的線程超時(shí)，則當(dāng)前線程拋出 BrokenBarrierException 異常，并停止等待，繼續(xù)執(zhí)行。
- 其他線程調(diào)用 CyclicBarrier.reset() 方法，則當(dāng)前線程拋出 BrokenBarrierException 異常，并停止等待，繼續(xù)執(zhí)行。
- reset 方法：使得CyclicBarrier回歸初始狀態(tài)，直觀來看它做了兩件事：
- 如果有正在等待的線程，則會(huì)拋出 BrokenBarrierException 異常，且這些線程停止等待，繼續(xù)執(zhí)行。
- 將是否破損標(biāo)志位 broken 置為 false。
優(yōu)缺點(diǎn)分析CyclicBarrier 從設(shè)計(jì)的復(fù)雜度到使用的復(fù)雜度都高于 CountDownLatch，相比于 CountDownLatch 來說它的優(yōu)點(diǎn)是可以重復(fù)使用（只需調(diào)用 reset 就能恢復(fù)到初始狀態(tài)），缺點(diǎn)是使用難度較高。
小結(jié)
在實(shí)現(xiàn)判斷線程池任務(wù)是否執(zhí)行完成的方案中，通過統(tǒng)計(jì)線程池執(zhí)行完任務(wù)的方式（實(shí)現(xiàn)方法 1），以及實(shí)現(xiàn)方法 3（CountDownLatch 或 CyclicBarrier）等統(tǒng)計(jì)，都是“不記名”的，只關(guān)注數(shù)量，不關(guān)注（具體）對(duì)象，所以這些方式都有可能受到外界代碼的影響，因此使用 FutureTask 等待具體任務(wù)執(zhí)行完的方式是最推薦的判斷方法。
