文章目录
- 前言
- 1.需求分析
- 2.实现概述
- 3.具体实现
- 3.1 统计完成任务数
- 3.2 FutureTask
- 3.3 CountDownLatch和CyclicBarrier
- 小结
前言
论是在项目开发中，还是在面试中过程中，总会被问到或使用到并发编程来完成项目中的某个功能。
例如某个复杂的查询，无法使用一个查询语句来完成此功能，此时我们就需要执行多个查询语句，然后再将各自查询的结果，组装之后返回给前端了，那么这种场景下，我们就必须使用线程池来进行并发查询了。
PS：磊哥做的最复杂的查询，总共关联了 21 张表，在和产品及需求方的沟通多次沟通下，才将查询的业务从 21 张表，降到了至少要查询 12 张表（非常难搞），那么这种场景下是无法使用一个查询语句来实现的，那么并发查询是必须要给安排上的。
1.需求分析
线程池的使用并不复杂，麻烦的是如何判断线程池中的任务已经全部执行完了？因为我们要等所有任务都执行完之后，才能进行数据的组装和返回，所以接下来，我们就来看如何判断线程中的任务是否已经全部执行完？
2.实现概述
判断线程池中的任务是否执行完的方法有很多，比如以下几个：
使用 getCompletedTaskCount() 统计已经执行完的任务，和 getTaskCount() 线程池的总任务进行对比，如果相等则说明线程池的任务执行完了，否则既未执行完。
使用 FutureTask 等待所有任务执行完，线程池的任务就执行完了。
使用 CountDownLatch 或 CyclicBarrier 等待所有线程都执行完之后，再执行后续流程。
具体实现代码如下。
3.具体实现
3.1 统计完成任务数
通过判断线程池中的计划执行任务数和已完成任务数，来判断线程池是否已经全部执行完，如果计划执行任务数=已完成任务数，那么线程池的任务就全部执行完了，否则就未执行完。
示例代码如下：
private static void isCompletedByTaskCount(ThreadPoolExecutor threadPool) {
while (threadPool.getTaskCount() != threadPool.getCompletedTaskCount()) {
}
}
以上程序执行结果如下：
方法说明
getTaskCount()：返回计划执行的任务总数。由于任务和线程的状态可能在计算过程中动态变化，因此返回的值只是一个近似值。
getCompletedTaskCount()：返回完成执行任务的总数。因为任务和线程的状态可能在计算过程中动态地改变，所以返回的值只是一个近似值，但是在连续的调用中并不会减少。
缺点分析
此判断方法的缺点是 getTaskCount() 和 getCompletedTaskCount() 返回的是一个近似值，因为线程池中的任务和线程的状态可能在计算过程中动态变化，所以它们两个返回的都是一个近似值。
3.2 FutureTask
FutrueTask 的优势是任务判断精准，调用每个 FutrueTask 的 get 方法就是等待该任务执行完，如下代码所示：
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
/**
* 使用 FutrueTask 等待线程池执行完全部任务
*/
public class FutureTaskDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 创建一个固定大小的线程池
ExecutorService executor = Executors.newFixedThreadPool(3);
// 创建任务
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;
});
// 提交三个任务给线程池
executor.submit(task1);
executor.submit(task2);
executor.submit(task3);
// 等待所有任务执行完毕并获取结果
int result1 = task1.get();
int result2 = task2.get();
int result3 = task3.get();
System.out.println("Do main thread.");
}
}
以上程序的执行结果如下：
3.3 CountDownLatch和CyclicBarrier
CountDownLatch 和 CyclicBarrier 类似，都是等待所有任务到达某个点之后，再进行后续的操作，如下图所示：
CountDownLatch 使用的示例代码如下：
public static void main(String[] args) throws InterruptedException {
// 创建线程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20,
0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024));
final int taskCount = 5; // 任务总数
// 单次计数器
CountDownLatch countDownLatch = new CountDownLatch(taskCount); // ①
// 添加任务
for (int i = 0; i < taskCount; i++) {
final int finalI = i;
threadPool.submit(new Runnable() {
@Override
public void run() {
try {
// 随机休眠 0-4s
int sleepTime = new Random().nextInt(5);
TimeUnit.SECONDS.sleep(sleepTime);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(String.format("任务%d执行完成", finalI));
// 线程执行完，计数器 -1
countDownLatch.countDown(); // ②
}
});
}
// 阻塞等待线程池任务执行完
countDownLatch.await(); // ③
// 线程池执行完
System.out.println();
System.out.println("线程池任务执行完成！");
}
代码说明：以上代码中标识为 ①、②、③ 的代码行是核心实现代码，其中：
① 是声明一个包含了 5 个任务的计数器；
② 是每个任务执行完之后计数器 -1；
③ 是阻塞等待计数器 CountDownLatch 减为 0，表示任务都执行完了，可以执行 await 方法后面的业务代码了。
以上程序的执行结果如下：
缺点分析
CountDownLatch 缺点是计数器只能使用一次，CountDownLatch 创建之后不能被重复使用。
CyclicBarrier 和 CountDownLatch 类似，它可以理解为一个可以重复使用的循环计数器，CyclicBarrier 可以调用 reset 方法将自己重置到初始状态，CyclicBarrier 具体实现代码如下：
public static void main(String[] args) throws InterruptedException {
// 创建线程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20,
0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1024));
final int taskCount = 5; // 任务总数
// 循环计数器 ①
CyclicBarrier cyclicBarrier = new CyclicBarrier(taskCount, new Runnable() {
@Override
public void run() {
// 线程池执行完
System.out.println();
System.out.println("线程池所有任务已执行完！");
}
});
// 添加任务
for (int i = 0; i < taskCount; i++) {
final int finalI = i;
threadPool.submit(new Runnable() {
@Override
public void run() {
try {
// 随机休眠 0-4s
int sleepTime = new Random().nextInt(5);
TimeUnit.SECONDS.sleep(sleepTime);
System.out.println(String.format("任务%d执行完成", finalI));
// 线程执行完
cyclicBarrier.await(); // ②
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
});
}
}
以上程序的执行结果如下：
方法说明
CyclicBarrier 有 3 个重要的方法：
构造方法：构造方法可以传递两个参数，参数 1 是计数器的数量 parties，参数 2 是计数器为 0 时，也就是任务都执行完之后可以执行的事件（方法）。
await 方法：在 CyclicBarrier 上进行阻塞等待，当调用此方法时 CyclicBarrier 的内部计数器会 -1，直到发生以下情形之一：
在 CyclicBarrier 上等待的线程数量达到 parties，也就是计数器的声明数量时，则所有线程被释放，继续执行。
当前线程被中断，则抛出 InterruptedException 异常，并停止等待，继续执行。
其他等待的线程被中断，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。
其他等待的线程超时，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。
其他线程调用 CyclicBarrier.reset() 方法，则当前线程抛出 BrokenBarrierException 异常，并停止等待，继续执行。
reset 方法：使得CyclicBarrier回归初始状态，直观来看它做了两件事：
如果有正在等待的线程，则会抛出 BrokenBarrierException 异常，且这些线程停止等待，继续执行。
将是否破损标志位 broken 置为 false。
优缺点分析
CyclicBarrier 从设计的复杂度到使用的复杂度都高于 CountDownLatch，相比于 CountDownLatch 来说它的优点是可以重复使用（只需调用 reset 就能恢复到初始状态），缺点是使用难度较高。
小结
在实现判断线程池任务是否执行完成的方案中，通过统计线程池执行完任务的方式（实现方法 1），以及实现方法 3（CountDownLatch 或CyclicBarrier）等统计，都是“不记名”的，只关注数量，不关注（具体）对象，所以这些方式都有可能受到外界代码的影响，因此使用 FutureTask 等待具体任务执行完的方式是最推荐的判断方法。
