5、线程池
作者：互联网
文章目录
一、ThreadPoolExecutor
ThreadPoolExecutor类，线程池的核心类
1、ThreadPoolExecutor类
-
构造参数：
- corePoolSize:核心池的大小，当线程数达到corePoolSize的时候，就会把到达的任务放到缓存队列
- maximumPoolSize:线程池最大线程数
- keepAliveTime:线程没有执行任务时最多保持多久时间会终止
- unit:keepAliveTime的时间单位，共有七种，天、小时、分钟、秒钟、毫秒、微秒、纳秒
- workQueue:一个阻塞队列，存储等待执行的任务
- threadFactory:线程工厂，用来创建线程
- hander:表示当拒绝处理任务时的策略
-
实现继承关系
-
重要方法
- execute():核心方法，Executor声明，ThreadPoolExecutor实现，通过这个方法向线程池提交一个任务，交由线程池去执行
- submit():ExecutorService中声明的方法，在AbstractExecutorService中实现。用来向线程池提交任务，但是与executor()不同，它能返回任务执行的结果，submit实际上还调用executor()方法
- shutdown()
- shutdownNow()
shutdown()和shutdownNow()
都用来关闭线程池的。
二、深入剖析线程池原理
1 线程池的状态
volatile int runState;
static final int RUNNING=0;
static final int SHUTDOWN=1;
static final int STOP=2;
static final int TERMINATED=3;
runState表示当前线程池的状态，它时一个volatile变量用来保证线程之间的可见性；
-
当创建线程池后，初始时，线程池处于RUNNING状态；
-
如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕；
-
如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务；
-
当线程池处于SHUTDOWN或STOP状态，并且所有工作线程已经销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATED状态。
2 任务执行
- execute()方法的实现原理
public void execute(Runnable command){
if(command==null)
throw new NullPointException();
//如果线程数量大于等于核心池数量 或者 addIfUnderCorePoolSize这个方法返回false
if(poolSize>=corePoolSize ||!addIfUnderCorePoolSize(command)){
//如果线程池状态为RUNNING状态则将任务放入任务缓存队列
//如果线程池状态不为RUNNING状态或者任务放入任务缓存队列
if(runState==RUNNING &&workQueue.offer(command)){
//为了防止在将此任务添加进任务缓存队列的同时，其他线程突然调用shutdowm
if(runState!=RUNNING || poolSize==0)
//保证添加到任务缓存队列中的任务得到处理
ensureQueuedTaskHandled(command);
}
//如果执行addIfUnderMaximumPoolSize方法失败，则执行reject()方法进行任务拒绝处理
else if(!addIfUnderMaximumPoolSize(command))
reject(command);
}
}
- addIfUnderCorePoolSize方法
private boolean addIfUnderCorePoolSize(Runnable firstTask){
Thread t=null;
//ReentrantLock锁
final ReentrantLock mainLock=this.mainLock;
mainLock.lock();
try{
if(poolSize<corePoolSize && runState==RUNNING)
//创建线程去执行firstTask任务
t=addThread(firstTask);
}finally{
mainLock.unlock();
}
if(t==null)
return false;
t.start();
return true;
}
首先获取到锁，因为这地方涉及到线程池状态的变化，先通过if语句判断当前线程池中的线程数目是否小于核心池大小，有朋友也许会有疑问：前面在execute()方法中不是已经判断过了吗，只有线程池当前线程数目小于核心池大小才会执行addIfUnderCorePoolSize方法的，为何这地方还要继续判断？原因很简单，前面的判断过程中并没有加锁，因此可能在execute方法判断的时候poolSize小于corePoolSize，而判断完之后，在其他线程中又向线程池提交了任务，就可能导致poolSize不小于corePoolSize了，所以需要在这个地方继续判断。然后接着判断线程池的状态是否为RUNNING，原因也很简单，因为有可能在其他线程中调用了shutdown或者shutdownNow方法。
然后就是执行t = addThread(firstTask);这个方法也非常关键，传进去的参数为提交的任务，返回值为Thread类型。然后接着在下面判断t是否为空，为空则表明创建线程失败（即poolSize>=corePoolSize或者runState不等于RUNNING），否则调用t.start()方法启动线程。
- addThread方法
private Thread addThread(Runnable firstTask){
Worker w=new worker(firstTask);
//创建一个线程，执行任务
Thread t=threadFactory.newThread(w);
if(t!=null){
//将创建的线程的引用赋值为w的成员变量
w.thread=t;
workers.add(w);
//当前线程数加1
int nt=++poolSize;
if(nt>largestPoolSize)
largestPoolSize=nt;
}
return t;
}
在addThread方法中，首先用提交的任务创建了一个Worker对象，然后调用线程工厂threadFactory创建了一个新的线程t，然后将线程t的引用赋值给了Worker对象的成员变量thread，接着通过workers.add(w)将Worker对象添加到工作集当中。
- Work类的实现
private final class Worker implements Runnable{
private final ReentrantLock runLock =new ReentrantLock();
private Runnable firstTask;
volatile long completedTasks;
Thread thread;
Woker(Runnable firstTask){
this.firstTask=first;
}
boolean isActive(){
return runLock.isLocked();
}
void interruptifidle(){
final ReentrantLock runLock =this.runLock;
if(runLock.tryLock()){
try{
if(thread!=Thread.currentThread())
thread.interrupt();
}finally{
runLock.unlock();
}
}
}
void interruptNow() {
thread.interrupt();
}
private void runTask(Runnable task) {
final ReentrantLock runLock = this.runLock;
runLock.lock();
try {
if (runState < STOP &&
Thread.interrupted() &&
runState >= STOP)
boolean ran = false;
beforeExecute(thread, task); //beforeExecute方法是ThreadPoolExecutor类的一个方法，没有具体实现，用户可以根据
//自己需要重载这个方法和后面的afterExecute方法来进行一些统计信息，比如某个任务的执行时间等
try {
task.run();
ran = true;
afterExecute(task, null);
++completedTasks;
} catch (RuntimeException ex) {
if (!ran)
afterExecute(task, ex);
throw ex;
}
} finally {
runLock.unlock();
}
}
public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this); //当任务队列中没有任务时，进行清理工作
}
}
}
它实际上实现了Runnable接口，因此上面的Thread t = threadFactory.newThread(w);效果跟下面这句的效果基本一样：
Thread t = new Thread(w);
相当于传进去了一个Runnable任务，在线程t中执行这个Runnable。
既然Worker实现了Runnable接口，那么自然最核心的方法便是run()方法了：
public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}
从run方法的实现可以看出，它首先执行的是通过构造器传进来的任务firstTask，在调用runTask()执行完firstTask之后，在while循环里面不断通过getTask()去取新的任务来执行，那么去哪里取呢？自然是从任务缓存队列里面去取，getTask是ThreadPoolExecutor类中的方法，并不是Worker类中的方法，下面是getTask方法的实现：
Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN) // Help drain queue
r = workQueue.poll();
else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间，
//则通过poll取任务，若等待一定的时间取不到任务，则返回null
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take();
if (r != null)
return r;
if (workerCanExit()) { //如果没取到任务，即r为null，则判断当前的worker是否可以退出
if (runState >= SHUTDOWN) // Wake up others
interruptIdleWorkers(); //中断处于空闲状态的worker
return null;
}
// Else retry
} catch (InterruptedException ie) {
// On interruption, re-check runState
}
}
}
在getTask中，先判断当前线程池状态，如果runState大于SHUTDOWN（即为STOP或者TERMINATED），则直接返回null。
如果runState为SHUTDOWN或者RUNNING，则从任务缓存队列取任务。
如果当前线程池的线程数大于核心池大小corePoolSize或者允许为核心池中的线程设置空闲存活时间，则调用poll(time,timeUnit)来取任务，这个方法会等待一定的时间，如果取不到任务就返回null。
然后判断取到的任务r是否为null，为null则通过调用workerCanExit()方法来判断当前worker是否可以退出，我们看一下workerCanExit()的实现：
private boolean workerCanExit() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
boolean canExit;
//如果runState大于等于STOP，或者任务缓存队列为空了
//或者 允许为核心池线程设置空闲存活时间并且线程池中的线程数目大于1
try {
canExit = runState >= STOP ||
workQueue.isEmpty() ||
(allowCoreThreadTimeOut &&
poolSize > Math.max(1, corePoolSize));
} finally {
mainLock.unlock();
}
return canExit;
}
也就是说如果线程池处于STOP状态、或者任务队列已为空或者允许为核心池线程设置空闲存活时间并且线程数大于1时，允许worker退出。如果允许worker退出，则调用interruptIdleWorkers()中断处于空闲状态的worker，我们看一下interruptIdleWorkers()的实现：
void interruptIdleWorkers() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) //实际上调用的是worker的interruptIfIdle()方法
w.interruptIfIdle();
} finally {
mainLock.unlock();
}
}
从实现可以看出，它实际上调用的是worker的interruptIfIdle()方法，在worker的interruptIfIdle()方法中：
void interruptIfIdle() {
final ReentrantLock runLock = this.runLock;
if (runLock.tryLock()) { //注意这里，是调用tryLock()来获取锁的，因为如果当前worker正在执行任务，锁已经被获取了，是无法获取到锁的
//如果成功获取了锁，说明当前worker处于空闲状态
try {
if (thread != Thread.currentThread())
thread.interrupt();
} finally {
runLock.unlock();
}
}
}
这里有一个非常巧妙的设计方式，假如我们来设计线程池，可能会有一个任务分派线程，当发现有线程空闲时，就从任务缓存队列中取一个任务交给空闲线程执行。但是在这里，并没有采用这样的方式，因为这样会要额外地对任务分派线程进行管理，无形地会增加难度和复杂度，这里直接让执行完任务的线程去任务缓存队列里面取任务来执行。
我们再看addIfUnderMaximumPoolSize方法的实现，这个方法的实现思想和addIfUnderCorePoolSize方法的实现思想非常相似，唯一的区别在于addIfUnderMaximumPoolSize方法是在线程池中的线程数达到了核心池大小并且往任务队列中添加任务失败的情况下执行的：
private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < maximumPoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
看到没有，其实它和addIfUnderCorePoolSize方法的实现基本一模一样，只是if语句判断条件中的poolSize < maximumPoolSize不同而已。
到这里，大部分朋友应该对任务提交给线程池之后到被执行的整个过程有了一个基本的了解，下面总结一下：
1）首先，要清楚corePoolSize和maximumPoolSize的含义；
2）其次，要知道Worker是用来起到什么作用的；
3）要知道任务提交给线程池之后的处理策略，这里总结一下主要有4点：
如果当前线程池中的线程数目小于corePoolSize，则每来一个任务，就会创建一个线程去执行这个任务；
如果当前线程池中的线程数目>=corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列当中，若添加成功，则该任务会等待空闲线程将其取出去执行；若添加失败（一般来说是任务缓存队列已满），则会尝试创建新的线程去执行这个任务；
如果当前线程池中的线程数目达到maximumPoolSize，则会采取任务拒绝策略进行处理；
如果线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止，直至线程池中的线程数目不大于corePoolSize；如果允许为核心池中的线程设置存活时间，那么核心池中的线程空闲时间超过keepAliveTime，线程也会被终止。
3 线程池中的线程初始化
默认情况下，创建线程池之后，线程池中
[外链图片转存失败(img-iLCLz7zW-1563932597078)(en-resource://database/661:1)]
标签：Runnable,任务,线程,firstTask,null,方法 来源： https://blog.csdn.net/dxx707099957/article/details/97103860