飞道的博客

Java线程池深度揭秘

462人阅读  评论(0)

作为 Java 程序员,无论是技术面试、项目研发或者是学习框架源码,不彻底掌握 Java 多线程的知识,做不到心中有数,干啥都没底气,尤其是技术深究时往往略显发憷。

坐稳扶好,通过今天的分享,能让你轻松 get 如下几点。

1. Executor 框架家族简介;

2. 源码解读:线程池状态以及状态流转;

3. 源码解读:部分成员变量及方法;

4. 源码解读:任务提交submit方法背后;

5. 源码揭秘之后的反思;

6. 寄语。

1. Excutor 家族简介  

一图胜千言,脑中有图心不慌。

(一)Executor 接口。


   
  1. public interface Executor {
  2. void execute(Runnable command);
  3. }

Executor 是一个接口(主要用于定义规范),定义了 execute 方法,用于接收 Runnable 对象。

(二)ExecutorService 接口。


   
  1. public interface ExecutorService extends Executor {
  2. // ... ...
  3. <T> Future<T> submit(Callable<T> task);
  4. <T> Future<T> submit(Runnable task, T result);
  5. Future<?> submit(Runnable task);
  6. // ... ...
  7. }

ExecutorService 也是一个接口,继承了 Executor 接口,增加了更多方法,相当于扩展了 Executor 接口的功能,例如定义了 submit() 系列方法,支持任务执行后得到返回结果。

(三)AbstractExecutorService 抽象类。


   
  1. public abstract class AbstractExecutorService implements ExecutorService {
  2. // ... ...
  3. protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
  4. return new FutureTask<T>(runnable, value);
  5. }
  6. protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
  7. return new FutureTask<T>(callable);
  8. }
  9. public Future<?> submit(Runnable task) {
  10. if (task == null) throw new NullPointerException();
  11. RunnableFuture<Void> ftask = newTaskFor(task, null);
  12. execute(ftask);
  13. return ftask;
  14. }
  15. public <T> Future<T> submit(Runnable task, T result) {
  16. if (task == null) throw new NullPointerException();
  17. RunnableFuture<T> ftask = newTaskFor(task, result);
  18. execute(ftask);
  19. return ftask;
  20. }
  21. public <T> Future<T> submit(Callable<T> task) {
  22. if (task == null) throw new NullPointerException();
  23. RunnableFuture<T> ftask = newTaskFor(task);
  24. execute(ftask);
  25. return ftask;
  26. }
  27. // ... ...
  28. }

AbstractExecutorService 是一个抽象类,实现了 ExecutorService 接口中的部分方法,例如提供了任务提交的 submit 方法的默认实现,而 submit 方法最终会调用 execute 方法。

不过 AbstractExecutorService 并没有实现 execute 方法,相当于为子类留了个口子,让子类去灵活扩展(钩子函数)。

(四)ScheduledExecutorService 接口。


   
  1. public interface ScheduledExecutorService extends ExecutorService {
  2. public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
  3. public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
  4. public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
  5. public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
  6. }

ScheduledExecutorService 接口继承了 ExecutorService,增加定时调度的方法,使其成为一个可定时调度任务的接口,相当于扩展了 ExecutorService 的功能。

(五)ScheduledThreadPoolExecutor 类。


   
  1. public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {
  2. // ... ...
  3. }

ScheduledThreadPoolExecutor 类继承自 ThreadPoolExecutor 类,并且实现了 ScheduledExecutorService 接口,变成一个可定时调度任务的线程池。

(六)ThreadPoolExecutor 类。


   
  1. public class ThreadPoolExecutor extends AbstractExecutorService {
  2. // ... ...
  3. }

ThreadPoolExecutor 继承 AbstractExecutorService 抽象类,并实现了 execute 等一系列方法。

(七)Executors 类。


   
  1. public class Executors {
  2. // ... ...
  3. public static ExecutorService newFixedThreadPool(int nThreads) {
  4. return new ThreadPoolExecutor(nThreads, nThreads,
  5. 0L, TimeUnit.MILLISECONDS,
  6. new LinkedBlockingQueue<Runnable>());
  7. }
  8. // ... ...
  9. }

研发人员可以通过 Executors 工厂类来创建线程池并返回一个ExecutorService 对象,而内部几乎全是对 ThreadPoolExecutor 的封装。

通过 Executor 的家族简单认识,应该能感觉到 ThreadPoolExecutor 类的重要性,所以接下来要重点对 ThreadPoolExecutor 类的源码进行剖析。

2. 源码解读:线程池状态以及状态流转  

上面注释截图来源于 ThreadPoolExecutor 的源码,别懵圈,仔细看差不多都能懂,能够看出线程池的五种状态以及对应的状态流转。

不知道你能看懂多少,看不懂也没关系,接下来把上面的注释用图呈现给大家。通过源码中的注释,能够勾勒出如下线程池的状态流转图(好的注释是多么的重要啊,感叹号!)。

3. 源码解读部分成员变量及方法  


   
  1. /**
  2. * ctl 是一个 AtomicInteger 类型的原子对象。
  3. * 其实设计很有意思:ctl 共包括 32 位(高 3 位表示"线程池状态",低 29 位表示"线程池中的线程数量")。
  4. * 个人感觉:线程池状态与线程数量合二为一,用一个变量来表示,来减少锁竞争,提高并发效率。
  5. */
  6. private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  7. /** 表示线程池线程数的位数:32 - 3 = 29 位 */
  8. private static final int COUNT_BITS = Integer.SIZE - 3;
  9. /** 表示最大线程容量(000,11111111111111111111111111111)*/
  10. private static final int CAPACITY = ( 1 << COUNT_BITS) - 1;
  11. // runState is stored in the high-order bits(运行状态保存在 int 值的高 3 位)
  12. /** 111,00000000000000000000000000000 */
  13. private static final int RUNNING = -1 << COUNT_BITS;
  14. /** 000,00000000000000000000000000000 */
  15. private static final int SHUTDOWN = 0 << COUNT_BITS;
  16. /** 001,00000000000000000000000000000 */
  17. private static final int STOP = 1 << COUNT_BITS;
  18. /** 010,00000000000000000000000000000 */
  19. private static final int TIDYING = 2 << COUNT_BITS;
  20. /** 011,00000000000000000000000000000 */
  21. private static final int TERMINATED = 3 << COUNT_BITS;
  22. // Packing and unpacking ctl
  23. /** 获取线程池的运行状态 */
  24. private static int runStateOf(int c) { return c & ~CAPACITY; }
  25. /** 线程池内有效线程的数量 (workerCount) */
  26. private static int workerCountOf(int c) { return c & CAPACITY; }
  27. /** 线程池的状态和线程的数量组装,成为 ctl */
  28. private static int ctlOf(int rs, int wc) { return rs | wc; }

仔细去看上面的代码,注释已经很清晰啦。重点关注 ctl 变量,这个变量将线程池自身状态和线程数量,融合在这一个变量中,其中高 3 位表示线程池状态,低 29 位表示线程池中的线程数量,这样在多线程环境下更易保证线程池自身状态和线程数量的统一,不得不佩服源代码作者 Doug Lea,可谓是设计甚妙!

4. 源码解读任务提交 submit 方法背后  

疑问?当调用 submit() 方法,把一个任务提交给线程池去处理的时候,线程池的处理过程是什么样的呢?

通过开篇对 Executor 的家族简介,能够看到 submit() 方法最终会调用 ThreadPoolExecutor 的 execute 方法,走进源码好好看看 execute 方法都做了啥?

重点关注源码中的注释(红框圈住部分),若看懂注释,提交任务时线程池对应的处理,也就懂了一半啦(感触:好的编码规范真的好重要,业务开发时,核心代码一定要有注释)。

若依然很懵逼,一图胜千言,那就继续上图吧。

了解上图的整体流程,再去看看源码就彻悟啦。


   
  1. public void execute(Runnable command) {
  2. //【Step 0. 如果任务为空则抛出 NPE 异常】
  3. if (command == null)
  4. throw new NullPointerException();
  5. int c = ctl. get();
  6. //【Step 1. 判断核心线程是否已满】
  7. // 1.1. 判断当前线程数是否已经达到核心线程数的限制
  8. if (workerCountOf(c) < corePoolSize) {
  9. // 1.2. 如果未达到核心线程数的限制,则会直接添加一个核心线程,并指定首次执行的任务,进行任务处理
  10. if (addWorker(command, true))
  11. return;
  12. // 1.3. 如果添加失败,则刷新线程池的状态和线程的数量对应的变量 ctl
  13. c = ctl. get();
  14. }
  15. //【Step 2. 判断阻塞队列是否已满】
  16. // 2.1. 检查线程池是否是运行状态,然后将任务添加到等待队列
  17. if (isRunning(c) && workQueue.offer(command)) {
  18. // 2.2. 任务成功添加到等待队列,再次刷新 ctl
  19. int recheck = ctl. get();
  20. // 2.3. 添加任务到等待队列成功后,如果线程池不是运行状态,则将刚添加的任务从队列移除并执行拒绝策略
  21. if (! isRunning(recheck) && remove(command))
  22. reject(command);
  23. // 2.4. 判断当前线程数量,如果线程数量为 0,则添加一个非核心线程,并且不指定首次执行任务
  24. else if (workerCountOf(recheck) == 0)
  25. addWorker( null, false);
  26. }
  27. //【Step 3. 判断最大线程数量是否已经达到】
  28. // 3.1. 添加非核心线程,指定首次执行任务,如果添加失败,执行异常策略
  29. else if (!addWorker(command, false))
  30. reject(command);
  31. }

结合注释去读代码,应该都能搞懂。很显然 execute 方法中,多处都调用了 addWorker 方法,接下来简单剖析一下 addWorker 方法。


   
  1. private boolean addWorker(Runnable firstTask, boolean core) {
  2. retry:
  3. for (;;) {
  4. int c = ctl.get();
  5. // ... ...
  6. for (;;) {
  7. // ... ...
  8. // 通过 CAS 自旋,增加线程数 +1,增加成功跳出双层循环,继续往下执行
  9. if (compareAndIncrementWorkerCount(c))
  10. break retry;
  11. // ... ...
  12. }
  13. }
  14. // 到这儿,说明已经成功的将线程数 +1 了,但是真正的线程还没有被添加
  15. boolean workerStarted = false;
  16. boolean workerAdded = false;
  17. Worker w = null;
  18. try {
  19. // 添加线程,Worker 是继承了 AQS,实现了 Runnable 接口的包装类
  20. w = new Worker(firstTask);
  21. final Thread t = w.thread;
  22. if (t != null) {
  23. final ReentrantLock mainLock = this.mainLock;
  24. mainLock.lock();
  25. try {
  26. // ... ...
  27. // 添加新增的 Worker
  28. workers.add(w);
  29. int s = workers.size();
  30. if (s > largestPoolSize)
  31. largestPoolSize = s;
  32. workerAdded = true;
  33. // ... ...
  34. } finally {
  35. mainLock.unlock();
  36. }
  37. if (workerAdded) {
  38. // 启动 Worker
  39. t.start();
  40. workerStarted = true;
  41. }
  42. }
  43. } finally {
  44. if (! workerStarted)
  45. addWorkerFailed(w);
  46. }
  47. return workerStarted;
  48. }

为了简明扼要,方法酌情进行了删减。addWorker 方法主要是通过双重 for 循环进行线程数 +1,然后创建 Worker,并进行添加到 HashSet<Worker> workers 列表中,然后调用 t.start() 启动 Worker。

那么接下来再一起看看 Worker 里面都做了啥?


   
  1. private final class Worker
  2. extends AbstractQueuedSynchronizer
  3. implements Runnable {
  4. // ... ...
  5. final Thread thread;
  6. Runnable firstTask;
  7. /**
  8. * 通过指定的 firstTask 任务创建 Worker 对象
  9. */
  10. Worker(Runnable firstTask) {
  11. setState(- 1);
  12. this.firstTask = firstTask;
  13. // 通过当前 Worker 对象创建对应的线程对象 t,
  14. // 所以调用 t.start() 时最终会调用 Worker 的 run 方法
  15. this.thread = getThreadFactory().newThread( this);
  16. }
  17. public void run() {
  18. // run 方法最终会调用 ThreadPoolExecutor 的 runWorker 方法
  19. runWorker( this);
  20. }
  21. // ... ...
  22. }

通过 Worker 的构造函数能够了解到,会通过创建的 Worker 对象去构建线程对象,当线程对象启动时最终会调用 runWorker 方法。


   
  1. final void runWorker(Worker w) {
  2. Thread wt = Thread.currentThread();
  3. // 取出需要执行的任务
  4. Runnable task = w.firstTask;
  5. w.firstTask = null;
  6. w.unlock(); // allow interrupts
  7. boolean completedAbruptly = true;
  8. try {
  9. // 如果 task 不是 null 或者去 workQueue 队列中取到待执行的任务不为 null
  10. while (task != null || (task = getTask()) != null) {
  11. // ... ...
  12. try {
  13. // 开始执行任务前的钩子方法
  14. beforeExecute(wt, task);
  15. Throwable thrown = null;
  16. try {
  17. task.run();
  18. // ... ...
  19. } finally {
  20. // 任务执行后的钩子方法
  21. afterExecute(task, thrown);
  22. }
  23. } finally {
  24. // ... ...
  25. }
  26. }
  27. completedAbruptly = false;
  28. } finally {
  29. // Worker 退出
  30. processWorkerExit(w, completedAbruptly);
  31. }
  32. }

runWorker 方法,首先会取出要执行的任务 task,如果为空则会调用 getTask 方法从任务队列中获取,然后调用任务对应的 run 方法进行执行,另外预置了 beforeExecute、afterExecute 两个钩子函数,让研发人员监控线程执行成为可能。

另外,线程池中的线程如何从队列中获取待执行的任务的呢?走进 getTask 方法看一看。


   
  1. private Runnable getTask() {
  2. boolean timedOut = false; // Did the last poll() time out?
  3. // 这块体现了:线程池的线程是复用的,通过循环去获取队列中的任务去执行。
  4. for (;;) {
  5.          int c = ctl.get();
  6.          // ... ...
  7. int wc = workerCountOf(c);
  8. // allowCoreThreadTimeOut: 是否允许核心线程超时.
  9. // 如果设置为 false,那么线程池在达到 corePoolSize 个工作线程之前,不会让闲置的工作线程退出。
  10.          boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  11.          // ... ...
  12. try {
  13. // 从 workQueue 队列中取待执行的任务,根据 timed 来选择等待时间
  14. Runnable r = timed ?
  15. workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  16. workQueue.take();
  17. if (r != null)
  18. return r;
  19. timedOut = true;
  20. } catch (InterruptedException retry) {
  21. timedOut = false;
  22. }
  23. }
  24. }

为了便于理解,源码做了部分删减。重点关注从任务队列中获取待执行任务的对象的方法调用:workQueue.poll()、workQueue.take() ,前者是移除并返回队列中的头部元素,如果队列为空,则返回 null,而后者是移除并返回队列中的头部元素,如果队列为空,则阻塞。

烟未灭,酒过半 ... ... 源码探讨就谈到这儿... ...

5. 源码揭秘之后的反思 

(一)钩子函数的使用场景

场景一:如上面自定义的 MyThreadPoolExecutor,可以让日志打印线程及线程数等等信息。意味着研发人员可以扩展 ThreadPoolExecutor,对钩子函数 beforeExecute、afterExecute 进行实现,进而可以知晓线程池内部的调度细节,可以有效进行监控,针对故障排查应该很有帮助。

场景二:

AbstractExecutorService 并没有实现 execute 方法,而是为子类 ThreadPoolExecutor 留了个口子,让子类去灵活扩展(钩子函数)。

仔细想想业务开发时,诸多的使用场景,何尝不是如此呢?

(二)线程池的 submit 方法与 execute 方法啥区别呢?

execute 方法,适用于不需要关注返回值的场景,只需要将线程丢到线程池中去执行就可以了。

而 submit() 方法,适用于需要关注返回值的场景,不过最终会调用 execute() 方法。

考虑到性能提升,如果不需要关注返回值,则建议直接调用 execute() 方法,因为那样会屏蔽很多中间调度。

(三)线程池状态与线程数量用一个 ctl 变量表示的好处?

线程池状态和线程数量合二为一,用一个原子变量来表示,来减少锁竞争,提高并发效率。

(四)清晰的注释是否有必要?

通过探秘源码,很多图都是根据源码注释勾勒出来的。可以看出清晰的注释,对于核心流程而言真的很重要,一定要养成良好的编码习惯,关键业务逻辑、核心流程,建议一定要写好注释,利人又利己,何乐而不为之。


(五)Executor 家族框架,若写基础框架时,是否有借鉴意义呢?

个人感觉很有借鉴意义,因为无论业务开发还是基础服务,总会看到类似模式框架的身影,总会有大牛模仿着造轮子,所以闲暇之余可以抽象一下。

6. 寄语写最后 

本次,主要对 Executor 家族进行了简单介绍,并着重对线程池背后的 ThreadPoolExecutor 类进行深度剖析,知其然知其所以然,希望对大家有帮助。

好了,本次就谈到这里,一起聊技术、谈业务、喷架构,少走弯路,不踩大坑。会持续输出原创精彩分享,敬请期待!

推荐阅读:

Executors功能如此强大,ThreadPoolExecutor功不可没(一)

彻底搞懂 Java 线程池,干啥都不再发憷

Java程序跑的快,全要靠线程带

fastjson的这些坑,你误入了没?

真实|技术人员该如何站好最后一班岗?

Java 8 的这些特性,你知道吗?

改掉这些坏习惯,还怕写不出健壮的代码?(一)

改掉这些坏习惯,还怕写不出优雅的代码?(二)

改掉这些坏习惯,还怕写不出优雅的代码?(三)

改掉这些坏习惯,还怕写不出健壮的代码?(四)

改掉这些坏习惯,还怕写不出精简的代码?(五)

改掉这些坏习惯,还怕写不出精简的代码?(六)


转载:https://blog.csdn.net/javaforwork/article/details/107829066
查看评论
* 以上用户言论只代表其个人观点,不代表本网站的观点或立场