java.uitl.concurrent.ThreadPoolExecutor 类是线程池中最核心的一个类,因此如果要透彻地了解 Java 中的线程池,必须先了解这个类。下面我们来看一下 ThreadPoolExecutor 类的具体实现源码。
在 ThreadPoolExecutor 类中提供了四个构造方法:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
|
从上面的代码可以得知,ThreadPoolExecutor 继承了 AbstractExecutorService 类,并提供了四个构造器,事实上,通过观察每个构造器的源码具体实现,发现前面三个构造器都是调用的第四个构造器进行的初始化工作。
下面解释下一下构造器中各个参数的含义:
- TimeUnit.DAYS; //天
- TimeUnit.HOURS; //小时
- TimeUnit.MINUTES; //分钟
- TimeUnit.SECONDS; //秒
- TimeUnit.MILLISECONDS; //毫秒
- TimeUnit.MICROSECONDS; //微妙
- TimeUnit.NANOSECONDS; //纳秒
- ArrayBlockingQueue;
- LinkedBlockingQueue;
- SynchronousQueue;
ArrayBlockingQueue 和 PriorityBlockingQueue 使用较少,一般使用 LinkedBlockingQueue 和 Synchronous。线程池的排队策略与 BlockingQueue 有关。
- ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
- ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。
- ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
- ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务
具体参数的配置与线程池的关系将在下一节讲述。
从上面给出的 ThreadPoolExecutor 类的代码可以知道,ThreadPoolExecutor 继承了 AbstractExecutorService,我们来看一下 AbstractExecutorService 的实现:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
|
AbstractExecutorService 是一个抽象类,它实现了 ExecutorService 接口。
我们接着看 ExecutorService 接口的实现:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
|
而 ExecutorService 又是继承了 Executor 接口,我们看一下 Executor 接口的实现:
1 2 3 |
|
到这里,大家应该明白了 ThreadPoolExecutor、AbstractExecutorService、ExecutorService 和 Executor 几个之间的关系了。
Executor 是一个顶层接口,在它里面只声明了一个方法 execute(Runnable),返回值为 void,参数为 Runnable 类型,从字面意思可以理解,就是用来执行传进去的任务的;
然后 ExecutorService 接口继承了 Executor 接口,并声明了一些方法:submit、invokeAll、invokeAny 以及 shutDown 等;
抽象类 AbstractExecutorService 实现了 ExecutorService 接口,基本实现了 ExecutorService 中声明的所有方法;
然后 ThreadPoolExecutor 继承了类 AbstractExecutorService。
在 ThreadPoolExecutor 类中有几个非常重要的方法:
1 2 3 4 |
|
execute() 方法实际上是 Executor 中声明的方法,在 ThreadPoolExecutor 进行了具体的实现,这个方法是 ThreadPoolExecutor 的核心方法,通过这个方法可以向线程池提交一个任务,交由线程池去执行。
submit() 方法是在 ExecutorService 中声明的方法,在 AbstractExecutorService 就已经有了具体的实现,在 ThreadPoolExecutor 中并没有对其进行重写,这个方法也是用来向线程池提交任务的,但是它和 execute() 方法不同,它能够返回任务执行的结果,去看 submit() 方法的实现,会发现它实际上还是调用的 execute() 方法,只不过它利用了 Future 来获取任务执行结果(Future 相关内容将在下一篇讲述)。
shutdown() 和 shutdownNow() 是用来关闭线程池的。
还有很多其他的方法:
比如:getQueue() 、getPoolSize() 、getActiveCount()、getCompletedTaskCount() 等获取与线程池相关属性的方法,有兴趣的朋友可以自行查阅 API。
在上一节我们从宏观上介绍了 ThreadPoolExecutor,下面我们来深入解析一下线程池的具体实现原理,将从下面几个方面讲解:
1. 线程池状态
2. 任务的执行
3. 线程池中的线程初始化
4. 任务缓存队列及排队策略
5. 任务拒绝策略
6. 线程池的关闭
7. 线程池容量的动态调整
1. 线程池状态
在 ThreadPoolExecutor 中定义了一个 volatile 变量,另外定义了几个 static final 变量表示线程池的各个状态:
1 2 3 4 5 |
|
runState 表示当前线程池的状态,它是一个 volatile 变量用来保证线程之间的可见性;
下面的几个 static final 变量表示 runState 可能的几个取值。
当创建线程池后,初始时,线程池处于 RUNNING 状态;
如果调用了 shutdown() 方法,则线程池处于 SHUTDOWN 状态,此时线程池不能够接受新的任务,它会等待所有任务执行完毕;
如果调用了 shutdownNow() 方法,则线程池处于 STOP 状态,此时线程池不能接受新的任务,并且会去尝试终止正在执行的任务;
当线程池处于 SHUTDOWN 或 STOP 状态,并且所有工作线程已经销毁,任务缓存队列已经清空或执行结束后,线程池被设置为 TERMINATED 状态。
2. 任务的执行
在了解将任务提交给线程池到任务执行完毕整个过程之前,我们先来看一下 ThreadPoolExecutor 类中其他的一些比较重要成员变量:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
|
每个变量的作用都已经标明出来了,这里要重点解释一下 corePoolSize、maximumPoolSize、largestPoolSize 三个变量。
corePoolSize 在很多地方被翻译成核心池大小,其实我的理解这个就是线程池的大小。举个简单的例子:
假如有一个工厂,工厂里面有 10 个工人,每个工人同时只能做一件任务。
因此只要当 10 个工人中有工人是空闲的,来了任务就分配给空闲的工人做;
当 10 个工人都有任务在做时,如果还来了任务,就把任务进行排队等待;
如果说新任务数目增长的速度远远大于工人做任务的速度,那么此时工厂主管可能会想补救措施,比如重新招 4 个临时工人进来;
然后就将任务也分配给这 4 个临时工人做;
如果说着 14 个工人做任务的速度还是不够,此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。
当这 14 个工人当中有人空闲时,而新任务增长的速度又比较缓慢,工厂主管可能就考虑辞掉 4 个临时工了,只保持原来的 10 个工人,毕竟请额外的工人是要花钱的。
这个例子中的 corePoolSize 就是 10,而 maximumPoolSize 就是 14(10+4)。
也就是说 corePoolSize 就是线程池大小,maximumPoolSize 在我看来是线程池的一种补救措施,即任务量突然过大时的一种补救措施。
不过为了方便理解,在本文后面还是将 corePoolSize 翻译成核心池大小。
largestPoolSize 只是一个用来起记录作用的变量,用来记录线程池中曾经有过的最大线程数目,跟线程池的容量没有任何关系。
下面我们进入正题,看一下任务从提交到最终执行完毕经历了哪些过程。
在 ThreadPoolExecutor 类中,最核心的任务提交方法是 execute() 方法,虽然通过 submit 也可以提交任务,但是实际上 submit 方法里面最终调用的还是 execute() 方法,所以我们只需要研究 execute() 方法的实现原理即可:
1 2 3 4 5 6 7 8 9 10 11 12 |
|
上面的代码可能看起来不是那么容易理解,下面我们一句一句解释:
首先,判断提交的任务 command 是否为 null,若是 null,则抛出空指针异常;
接着是这句,这句要好好理解一下:
1 |
|
由于是或条件运算符,所以先计算前半部分的值,如果线程池中当前线程数不小于核心池大小,那么就会直接进入下面的 if 语句块了。
如果线程池中当前线程数小于核心池大小,则接着执行后半部分,也就是执行
1 |
|
如果执行完 addIfUnderCorePoolSize 这个方法返回 false,则继续执行下面的 if 语句块,否则整个方法就直接执行完毕了。
如果执行完 addIfUnderCorePoolSize 这个方法返回 false,然后接着判断:
1 |
|
如果当前线程池处于 RUNNING 状态,则将任务放入任务缓存队列;如果当前线程池不处于 RUNNING 状态或者任务放入缓存队列失败,则执行:
1 |
|
如果执行 addIfUnderMaximumPoolSize 方法失败,则执行 reject() 方法进行任务拒绝处理。
回到前面:
1 |
|
这句的执行,如果说当前线程池处于 RUNNING 状态且将任务放入任务缓存队列成功,则继续进行判断:
1 |
|
这句判断是为了防止在将此任务添加进任务缓存队列的同时其他线程突然调用 shutdown 或者 shutdownNow 方法关闭了线程池的一种应急措施。如果是这样就执行:
1 |
|
进行应急处理,从名字可以看出是保证 添加到任务缓存队列中的任务得到处理。
我们接着看 2 个关键方法的实现:addIfUnderCorePoolSize 和 addIfUnderMaximumPoolSize:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
|
这个是 addIfUnderCorePoolSize 方法的具体实现,从名字可以看出它的意图就是当低于核心吃大小时执行的方法。下面看其具体实现,首先获取到锁,因为这地方涉及到线程池状态的变化,先通过 if 语句判断当前线程池中的线程数目是否小于核心池大小,有朋友也许会有疑问:前面在 execute() 方法中不是已经判断过了吗,只有线程池当前线程数目小于核心池大小才会执行 addIfUnderCorePoolSize 方法的,为何这地方还要继续判断?原因很简单,前面的判断过程中并没有加锁,因此可能在 execute 方法判断的时候 poolSize 小于 corePoolSize,而判断完之后,在其他线程中又向线程池提交了任务,就可能导致 poolSize 不小于 corePoolSize 了,所以需要在这个地方继续判断。然后接着判断线程池的状态是否为 RUNNING,原因也很简单,因为有可能在其他线程中调用了 shutdown 或者 shutdownNow 方法。然后就是执行
1 |
|
这个方法也非常关键,传进去的参数为提交的任务,返回值为 Thread 类型。然后接着在下面判断 t 是否为空,为空则表明创建线程失败(即 poolSize>=corePoolSize 或者 runState 不等于 RUNNING),否则调用 t.start() 方法启动线程。
我们来看一下 addThread 方法的实现:
1 2 3 4 5 6 7 8 9 10 11 12 |
|
在 addThread 方法中,首先用提交的任务创建了一个 Worker 对象,然后调用线程工厂 threadFactory 创建了一个新的线程 t,然后将线程 t 的引用赋值给了 Worker 对象的成员变量 thread,接着通过 workers.add(w) 将 Worker 对象添加到工作集当中。
下面我们看一下 Worker 类的实现:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
|
它实际上实现了 Runnable 接口,因此上面的 Thread t = threadFactory.newThread(w); 效果跟下面这句的效果基本一样:
1 |
|
相当于传进去了一个 Runnable 任务,在线程 t 中执行这个 Runnable。
既然 Worker 实现了 Runnable 接口,那么自然最核心的方法便是 run() 方法了:
1 2 3 4 5 6 7 8 9 10 11 12 |
|
从 run 方法的实现可以看出,它首先执行的是通过构造器传进来的任务 firstTask,在调用 runTask() 执行完 firstTask 之后,在 while 循环里面不断通过 getTask() 去取新的任务来执行,那么去哪里取呢?自然是从任务缓存队列里面去取,getTask 是 ThreadPoolExecutor 类中的方法,并不是 Worker 类中的方法,下面是 getTask 方法的实现:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
|
来源: http://www.bubuko.com/infodetail-2024938.html