【Java多线程】自己实现一个简单的线程池(二)

太过爱你忘了你带给我的痛 2022-11-25 10:08 197阅读 0赞

### 文章目录 ###

*   *  前言
     *  自定义任务完成后的通知
     *   *  实现原理
     *  JDK 的实现
     *  实现带有返回值的线程
     *   *  实现原理
     *  异常处理
     *  总结

## 前言 ##

再看本文之前，强烈建议先查看上篇[【Java多线程】自己实现一个简单的线程池(一)][Java]该文章主要实现了线程池的以下功能

1.  线程池基本调度功能。
2.  线程池自动扩容缩容。
3.  队列缓存线程。
4.  关闭线程池。

**这些功能，最后也留下了三个待实现的 features 。**

1.  执行带有返回值的线程。
2.  异常处理怎么办？
3.  所有任务执行完怎么通知我？

## 自定义任务完成后的通知 ##

大家在用线程池的时候或多或少都会有这样的需求：

**线程池中的任务执行完毕后再通知主线程做其他事情**，比如一批任务都执行完毕后再执行下一波任务等等。

> 如: 往线程池中提交了 13 个任务，直到所有任务都执行完毕后再打印 `“所有任务执行完毕”`这个日志。  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70]  
> **执行结果如下:**  
> ![在这里插入图片描述][20200808102451922.gif]

为了简单的达到这个效果，我们可以在初始化线程池的时候传入一个`Nofify接口`的实现，这个接口就是用于任务完成之后的回调。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 1]  
Notify接口只有一个方法

public interface Notify  { 
        /** * 回调函数 */
        void notifyListen();
    }

### 实现原理 ###

**想要实现这个功能的关键是在什么时候回调这个接口的实现？**

*  仔细想想其实也简单：只要我们`记录提交到线程池中的任务` 及 `完成的数量`，他们`两者的差为 0`时就认为`线程池中的任务已执行完毕；`这时便可回调这个接口

**所以在往线程池中写入任务时我们原子记录任务数量+1(原子递增)**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 2]  
**为了并发安全的考虑，这里的计数器采用了`原子类 AtomicInteger 。`**

/** * 提交到线程池的任务总数计数器 */
        private AtomicInteger totalTask = new AtomicInteger();

**线程池中的任务执行完成后递减任务数量-1：(原子递减)**

> `每次任务执行完毕后判断计数器是否为 0`，为0时 表示线程池所有任务执行完毕；这时便可回调我们自定义的接口完成通知。  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 3]

## JDK 的实现 ##

在 jdk 中的`java.util.concurrent.ThreadPoolExecutor`中也有相关的 API ，只是用法不太一样，但本质原理都大同小异。

使用 ThreadPoolExecutor 的常规关闭流程如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 4]  
执行结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 5]

线程提交完毕后执行`shutdown()`关闭线程池，接着循环调用`awaitTermination()`方法，一旦任务全部执行完毕后则会返回 true 从而退出循环。

这两个方法的目的和原理如下：

1.  执行 `shutdown()`后会将线程池的状态置为`关闭状态`，这时将会`停止接收新的任`务同时会`等待队列中的任务全部执行完毕`后才真正关闭线程池。
2.  `awaitTermination`会`阻塞`直到`线程池所有任务执行完毕或者超时时间已到`。**(true表示所有任务执行完毕或者超时时间已到,false还有任务正在运行或者等待运行)**

**为什么要两个 api 结合一起使用呢？**  
主要还在最终的目的是：**所有线程执行完毕后再做某件事情，也就是在线程执行完毕之前其实主线程是需要被阻塞的。**

*  `shutdown()`执行后并不会阻塞，会立即返回，所有才需要后续`用循环不停的调用 awaitTermination()`，因为这个 api 才会阻塞线程。
 *  其实我们查看源码会发现，`ThreadPoolExecutor`中的`阻塞`依然也是`等待通知机制`的运用，只不过用的是 `LockSupport 的 API`而已。

## 实现带有返回值的线程 ##

如: **需要线程异步计算某些数据然后得到结果最终汇总使用。**

1.  首先`任务Worker`是不能实现 `Runnable`接口了,它的 `run()`是没有返回值的；所以我们改成实现一个 Callable 的接口,他的`call()`是可以带返回值的

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 6]  
2. 在提交任务时也稍作改动  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 7]

1.  执行任务的函数由`execute()` 换为了 `submit()`，同时`submit()`会返回一个返回值`Future`
2.  保存每次提交任务返回的`Future`
3.  循环获取提交的`Future.get()`方法获取返回值

执行结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 8]

### 实现原理 ###

再看具体实现之前先来思考下这样的功能如何实现？

1.  受限于 jdk 的线程 api 的规范，要执行一个线程不管是实现`runnable接口`还是`继承Thread类`，最终都是执行的 `run() 函数`。
2.  所以我们想要一个线程有返回值无非只能是`在执行 run() 函数时去调用一个有返回值的方法`，`再将这个返回值存放起来用于后续使用。`

**1.新建了一个 `Callable<T>` 的接口**（和\*`java.util.concurrent.Callable`\*一样），它的`call()`就是刚才提到的有返回值的方法，所以我们应当`在线程的 run() 函数`中去调用它。  
![在这里插入图片描述][20200810143537786.png]

**2.新建一个`Future`的接口，他的主要作用是`获取线程的返回值`，也就是 `将这个返回值存放起来用于后续使用`**  
![在这里插入图片描述][20200810144208539.png]

3.**创建一个 `FutureTask`，它`实现了 Future 接口`用于后续获取返回值。同时`实现了 Runnable 接口`会把自己变为一个线程。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 9]

> 在`FutureTask的 run()`函数中会调用刚才提到的具有返回值的 `call()`。

4.**结合`CustomThreadPool`的 `submit() 提交任务`和 `get() 获取返回值的源码`来看会更加理解这其中的门道。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 10]

`submit()`非常简单，将我们传进来的`Callable 对象`转换为一个`FutureTask 对象`，然后再调用之前的 `execute()`来丢进线程池（**后续的执行就和一个普通的线程进入线程池的流程一样**）。

> `FutureTask实现了Runnable接口` 本身也是线程，所以可以直接使用 `execute()`函数。

5.**而`future.get()` 的真正实现是`FutureTask`，所以我们直接看其中的源码就好。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 11]

1.  get() 在线程没有返回之前是一个`阻塞函数`，最终也是通过 `notify.wait()`使线程进入阻塞状态来实现的。
2.  使其从`wait()`中返回的条件必然是在`线程执行完毕拿到返回值的时候才进行唤醒。`  
    一旦`线程执行完毕`（ callable.call()）就会`唤醒 notify 对象`，这样 get 方法也就能返回了。

--------------------

**同理x `ThreadPoolExecutor` 中的原理也是类似，只不过它考虑的细节更多所以看起来很复杂，但精简代码后核心也就是这些。甚至最终使用的 api 看起来都是类似的：**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 12]

## 异常处理 ##

最后一个是一些新手`使用线程池很容易踩坑`的一个地方：那就是`异常处理`。  
比如类似于这样的场景：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 13]  
上图创建只有一个线程的线程池，当循环到第1000次时抛出一个`空指针异常`，但这个异常又没有被捕获结果`线程既没有继续运行同时线程池也没有退出，会一直卡在这里。`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 14]

当我们`dump 线程快照`会发现：

*  这时线程池中还有一个线程在运行，**通过线程名称会发现这是新创建的一个线程**（`之前是 Thread-0，现在是 Thread-1`）。
 *  它的线程状态为 `WAITING` ，通过`堆栈`发现是卡在了 CustomThreadPool.java:的140处的`task()`处。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 15]
 *  就是卡在了从`队列里获取任务的地方`，由于此时的任务队列是空的，所以他会一直阻塞在这里。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 16]
 *  其实在`线程池内部`会对`线程的运行捕获异常`，但它**并不会处理**，只是`用于标记是否执行成功；`
    
     *  一旦执行失败则会`回收掉当前异常的线程`，然后`重新创建一个新的 Worker 线程` **继续从队列里取任务然后执行**。所以最终才会卡在 `从队列中取任务处`。（**其实 ThreadPoolExecutor 的异常处理也是类似的**）  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 17]  
        接着上面finally后的代码  
        ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 18]

> **`所以我们在使用线程池时，其中的任务一定要做好异常处理。`**

## 总结 ##

这一波下来我觉得线程池搞清楚没啥问题了，总的来看它内部运用了非常多的多线程解决方案，比如：

*  ReentrantLock 重入锁来保证线程写入的并发安全。
 *  利用等待通知机制来实现线程间通信（线程执行结果、等待线程池执行完毕等）。

最后也学会了：

*  标准的线程池关闭流程。
 *  如何使用有返回值的线程。
 *  线程异常捕获的重要性。

**本文所有源码：** [点击这里][Link 1]

> [《Java动手撸源码》手写实现线程池][Java 1] 这也是一篇好文章

[Java]: https://blog.csdn.net/qq877728715/article/details/107632454
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70]: /images/20221124/16be7a85666a45728eb7c4f5d845778c.png
[20200808102451922.gif]: /images/20221124/3809a7a45e72485a8fc40e54c0475245.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 1]: /images/20221124/2b77e35590e14701a65505111117b631.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 2]: /images/20221124/dc5e8169010048358cc5fa0eef6bdd16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 3]: /images/20221124/ff40bc84e6c44e87a4b57f2a721c8b44.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 4]: /images/20221124/dae9c2e7ec9840c18ce3d6f89e26f4b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 5]: /images/20221124/e94cc3fd552248238f172c697b3705c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 6]: /images/20221124/68735ff563b14c9abf4ba47c05fede7e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 7]: /images/20221124/215addf7eb364230b3bbe9034d111838.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 8]: /images/20221124/a3fa38274b82430c96b2ca552c5ae1e8.png
[20200810143537786.png]: /images/20221124/2fba3140dc6a4edf9ee728bbf5198ded.png
[20200810144208539.png]: /images/20221124/2f1e14e859574fdfb0de4646b39a3cc3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 9]: /images/20221124/7d62c2b398a34019817d28a5ec6903a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 10]: /images/20221124/4466cc35377b4d44990adce71d9d2f70.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 11]: /images/20221124/4b4936d2a0c84b26a7fe0696369b29f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 12]: /images/20221124/0a9cb5cee35e4d3c91cb024eb352f86d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 13]: /images/20221124/daf23b51ad9d4e51ae8da2713dec3cdd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 14]: /images/20221124/142a6d6afec5464db9e7dc3313f3105a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 15]: /images/20221124/a4bee9c6dd2c4dee920f64086e1a8cc2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 16]: /images/20221124/e36b9b4fd81c4e409180a8dc82125ba7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 17]: /images/20221124/dfde702ad70d4d7995783648ed9e8835.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70 18]: /images/20221124/73cc9a76c885480885c4bb00d609a504.png
[Link 1]: https://gitee.com/oyjp0524/java-base-demo/tree/master/src/main/java/com/demo/concurrent
[Java 1]: https://blog.csdn.net/qq_34037358/article/details/113412745