Java 并发编程深入学习——线程池及其实现原理
Posted Freeguy88
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Java 并发编程深入学习——线程池及其实现原理相关的知识，希望对你有一定的参考价值。
Java线程池介绍
线程池，从字面含义来看，是指管理一组同构工作线程的资源池。线程池是与工作队列（work Queue）密切相关的，其中工作队列中保存了所有等待执行的任务。工作者线程（Work Thread）的任务很简单：从工作队列中获取一个任务，执行任务，然后返回线程池并等待下一个任务。
使用线程池的优势
在线程池中执行任务比为每个任务分配一个线程优势更多。主要体现在一下几个方面：
1.通过重用现有的线程而不是创建新的线程，可以再处理多个请求时分摊在线程创建和销毁过程中产生的巨大开销；
2. 当请求到达时，工作线程通常已经存在，因此不会由于等待创建线程而延迟任务的执行，从而提高了响应性。
3. 通过适当调整线程池的大小，可以创建足够多的线程以便使处理器保持忙碌状态，同时还可以防止过多线程相互竞争资源而是应用性能耗尽内存或失败。
线程池分类
Java 类库提供了一个灵活的线程池以及一些有用的默认配置。可以通过调用Executors中的静态工厂方法之一来创建一个线程池：
1.newFixedThreadPool
newFixedThreadPool将创建一个固定长度的线程池，每当提交一个任务时就创建一个线程，直到达到线程池的最大数量。
重载方法1：
public static ExecutorService newFixedThreadPool(int nThreads)
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
重载方法2：
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
特点：线程数量固定，线程处于空闲状态时，它们并不会被回收，除非线程池被关闭。当所有线程都处于活动状态时，新任务都会处于等待状态，直到有线程空闲出来。基本线程数等于最大线程数，没有超时机制，使用无界的队列保存等待执行的任务。
2.newCachedThreadPool
newCachedThreadPool将创建一个可缓存的线程池，如果线程池的当前规模超过了处理需求时，那么将回收空闲的线程，而当需求增加时，则可以添加新的线程，线程池的规模不存在任何限制。
重载方法1
public static ExecutorService newCachedThreadPool()
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
重载方法2
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory)
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
threadFactory);
特点：基本线程数为0，最大线程数为Integer.MAX_VALUE.存活时间60s,采用异步队列SynchronousQueue来避免任务排队。
3.newSingleThreadExecutor
newSingleThreadExecutor是一个单线程的Executor,它创建单个工作者线程来执行任务，如果这个线程异常结束，会创建另一个线程来替代。它能确保依照任务在对列中的顺序来串行执行（例如FIFO,LIFO,优先级）。
重载版本1
public static ExecutorService newSingleThreadExecutor()
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
重载版本2
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory));
特点：基本线程数和最大线程数都为1，无存活时间，采用无界的
LinkedBlockingQueue来保存等待执行的任务。
4.newScheduledThreadPool
newScheduledThreadPool创建一个固定长度的线程池，而且以延迟或定时的方式来执行任务，类似于Timer.
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
return new ScheduledThreadPoolExecutor(corePoolSize);
public ScheduledThreadPoolExecutor(int corePoolSize)
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue());
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory)
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
public static ScheduledExecutorService newScheduledThreadPool(
int corePoolSize, ThreadFactory threadFactory)
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
ThreadPoolExecutor
ThreadPoolExecutor是线程池的真正实现，它的构造方法提供了一系列参数来配置线程池。下面是ThreadPoolExecutor的一个比较常用的构造方法。
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue)各个参数含义如下：
1.corePoolSize
线程池核心线程数，默认情况下，核心线程会在线程池中一直存活，即使它们处于闲置状态。如果ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true,那么闲置的核心线程在等待新任务到来时会有超时策略，这个时间间隔由keepAliveTime所指定的时长后，核心线程就会被终止。
2.maximumPoolSize
线程池所能容纳的最大线程数，当活动线程达到这个数值后，后续的新任务将被阻塞。
3.keepAliveTime
非核心线程闲置时的超时时长，超过这个时长，非核心线程就会被回收。当ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true时，keepAliveTime同样会作用于非核心线程。
4.unit
keepAliveTime 参数的时间单位。
5.workQueue
执行前用于保持任务的队列。此队列仅保持由 execute 方法提交的 Runnable 任务。
6.threadFactory
执行程序创建新线程时使用的工厂。
7.handler
由于超出线程范围和队列容量而使执行被阻塞时所使用的处理程序。
ThreadPoolExecutor执行任务时遵循以下规则：
1. 如果线程池中的线程数量未达到核心线程的数量，那么会直接启动一个核心线程来执行任务；
2. 如果线程池中的线程数量已经达到或者超过核心线程的数量，那么任务会被插入到任务队列中排队等待执行；
3. 如果步骤2中无法将任务插入到任务队列中，这往往是由于任务队列已满，这个时候如果线程数量未达到线程池规定的最大值，那么会立刻启动一个非核心线程来执行。
4. 如果步骤3中线程数量已经达到线程池规定的最大值，那么就拒绝执行此任务，ThreadPoolExecutor 会调用RejectedExecutionHandler的rejectedExecution方法来通知调用者。
线程池实现分析
下面从源码的角度分析一下线程池的实现方式。
1.线程池状态
在ThreadPoolExecutor中定义了一个volatile变量，另外定义了几个static final变量表示线程池的各个状态：
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.任务的执行
在了解将任务提交给线程池到任务执行完毕整个过程之前，我们先来看一下ThreadPoolExecutor类中其他的一些比较重要成员变量：
private final BlockingQueue<Runnable> workQueue; //任务缓存队列，用来存放等待执行的任务
private final ReentrantLock mainLock = new ReentrantLock(); //线程池的主要状态锁，对线程池状态（比如线程池大小 、runState等）的改变都要使用这个锁
private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集
private volatile long keepAliveTime; //线程存活时间
private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间
private volatile int corePoolSize; //核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int maximumPoolSize; //线程池最大能容忍的线程数
private volatile int poolSize; //线程池中当前的线程数
private volatile RejectedExecutionHandler handler; //任务拒绝策略
private volatile ThreadFactory threadFactory; //线程工厂，用来创建线程
private int largestPoolSize; //用来记录线程池中曾经出现过的最大线程数
private long completedTaskCount; //用来记录已经执行完毕的任务个数下面我们进入正题，看一下任务从提交到最终执行完毕经历了哪些过程。
在ThreadPoolExecutor类中，最核心的任务提交方法是execute()方法，虽然通过submit也可以提交任务，但是实际上submit方法里面最终调用的还是execute()方法，所以我们只需要研究execute()方法的实现原理即可：
public void execute(Runnable command)
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))
if (runState == RUNNING && workQueue.offer(command))
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
上面的代码可能看起来不是那么容易理解，下面我们一句一句解释：
首先，判断提交的任务command是否为null，若是null，则抛出空指针异常；
接着是这句，这句要好好理解一下：
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))由于是或条件运算符，所以先计算前半部分的值，如果线程池中当前线程数不小于核心池大小，那么就会直接进入下面的if语句块了。
如果线程池中当前线程数小于核心池大小，则接着执行后半部分，也就是执行
addIfUnderCorePoolSize(command)如果执行完addIfUnderCorePoolSize这个方法返回false，则继续执行下面的if语句块，否则整个方法就直接执行完毕了。
如果执行完addIfUnderCorePoolSize这个方法返回false，然后接着判断：
if (runState == RUNNING && workQueue.offer(command))如果当前线程池处于RUNNING状态，则将任务放入任务缓存队列；如果当前线程池不处于RUNNING状态或者任务放入缓存队列失败，则执行：
addIfUnderMaximumPoolSize(command)如果执行addIfUnderMaximumPoolSize方法失败，则执行reject()方法进行任务拒绝处理。
回到前面：
if (runState == RUNNING && workQueue.offer(command))这句的执行，如果说当前线程池处于RUNNING状态且将任务放入任务缓存队列成功，则继续进行判断：
if (runState != RUNNING || poolSize == 0)这句判断是为了防止在将此任务添加进任务缓存队列的同时其他线程突然调用shutdown或者shutdownNow方法关闭了线程池的一种应急措施。如果是这样就执行：
ensureQueuedTaskHandled(command)进行应急处理，从名字可以看出是保证 添加到任务缓存队列中的任务得到处理。
我们接着看2个关键方法的实现：addIfUnderCorePoolSize和addIfUnderMaximumPoolSize：
private boolean addIfUnderCorePoolSize(Runnable firstTask)
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask); //创建线程去执行firstTask任务
finally
mainLock.unlock();
if (t == null)
return false;
t.start();
return true;
这个是addIfUnderCorePoolSize方法的具体实现，从名字可以看出它的意图就是当低于核心池大小时执行的方法。
下面看其具体实现，
首先获取到锁，因为这地方涉及到线程池状态的变化，先通过if语句判断当前线程池中的线程数目是否小于核心池大小，
为何这地方还要继续判断？原因很简单，前面的判断过程中并没有加锁，因此可能在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.thread = t; //将创建的线程的引用赋值为w的成员变量
workers.add(w);
int nt = ++poolSize; //当前线程数加1
if (nt > largestPoolSize)
largestPoolSize = nt;
return t;
在addThread方法中，首先用提交的任务创建了一个Worker对象，然后调用线程工厂threadFactory创建了一个新的线程t，然后将线程t的引用赋值给了Worker对象的成员变量thread，接着通过workers.add(w)将Worker对象添加到工作集当中。
下面我们看一下Worker类的实现：
private final class Worker implements Runnable
private final ReentrantLock runLock = new ReentrantLock();
private Runnable firstTask;
volatile long completedTasks;
Thread thread;
Worker(Runnable firstTask)
this.firstTask = firstTask;
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点：
1. 如果当前线程池中的线程数目小于corePoolSize，则每来一个任务，就会创建一个线程去执行这个任务；
2. 如果当前线程池中的线程数目>=corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列当中，若添加成功，则该任务会等待空闲线程将其取出去执行；若添加失败（一般来说是任务缓存队列已满），则会尝试创建新的线程去执行这个任务；
3. 如果当前线程池中的线程数目达到maximumPoolSize，则会采取任务拒绝策略进行处理；
4. 如果线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止，直至线程池中的线程数目不大于corePoolSize；如果允许为核心池中的线程设置存活时间，那么核心池中的线程空闲时间超过keepAliveTime，线程也会被终止。
3.线程池中的线程初始化
默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。
在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个方法办到：
prestartCoreThread()：初始化一个核心线程；
prestartAllCoreThreads()：初始化所有核心线程下面是这2个方法的实现：
public boolean prestartCoreThread()
return addIfUnderCorePoolSize(null); //注意传进去的参数是null
public int prestartAllCoreThreads()
int n = 0;
while (addIfUnderCorePoolSize(null))//注意传进去的参数是null
++n;
return n;
注意上面传进去的参数是null，根据第2小节的分析可知如果传进去的参数为null，则最后执行线程会阻塞在getTask方法中的
r = workQueue.take();- 1
即等待任务队列中有任务。
4.任务缓存队列及排队策略
在前面我们多次提到了任务缓存队列，即workQueue，它用来存放等待执行的任务。
workQueue的类型为BlockingQueue，通常可以取下面三种类型：
1）ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小；
2）LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE；
3）synchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务。
5.任务拒绝策略
当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略，通常有以下四种策略：
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务6.线程池的关闭
ThreadPoolExecutor提供了两个方法，用于线程池的关闭，分别是shutdown()和shutdownNow()，其中：
shutdown()：不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务7.线程池容量的动态调整
ThreadPoolExecutor提供了动态调整线程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()，
setCorePoolSize：设置核心池大小
setMaximumPoolSize：设置线程池最大能创建的线程数目大小当上述参数从小变大时，ThreadPoolExecutor进行线程赋值，还可能立即创建新的线程来执行任务。
参考资料
1.《android开发艺术探索》
2.《Java 并发编程实践》
3. Java并发编程：线程池的使用
文章来源：https://blog.csdn.net/ylyg050518/article/details/52461047
以上是关于Java 并发编程深入学习——线程池及其实现原理的主要内容，如果未能解决你的问题，请参考以下文章