一文带你搞定线程池原理

1.使用线程池的意义何在？

项目开发中，为了统一管理线程，并有效精准地进行排错，我们经常要求项目人员统一使用线程池去创建线程。因为我们是在受不了有些人动不动就去创建一个线程，使用的多了以后，一旦报错就只有一个线程报错信息，还是线程的共用信息，再加上如果你将异常吃了(捕获后不做处理)的情况下，这个错误。。。。em，我实在不知道去哪里排查，不然你换个人试试吧。

2.线程池的重要参数----你真的了解吗

publicThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {

corePoolSize:核心线程数。设置核心线程数的意义何在？通俗来讲核心线程数就是正式员工，需要长期坚守岗位，有任务就需要执行。

maximumPoolSize:最大线程池个数。设置最大线程池数量的意义何在？其实就是一个容错机制，当你的需要执行的线程个数已经爆满并且超过的时候，提供了一个容错机制，可以保证在短期内多余的任务正常执行。相当于就是临时工，临时过来执行任务，任务结束后就可以走了。

keepAliveTime：保活的时间。设置的意义何在？当线程任务无剧增的情况下，维持在正常提亮。你无需那么多临时工来执行任务，所以规定时间，临时工可以走人了，也即是除核心线程外的线程可以回收了。

TimeUnit：保活的时间单位。这个就不多赘述了。

BlockingQueue：阻塞队列。设置阻塞队列的意义何在？当所有核心线程都正在工作时，将其放入阻塞队列，等待后续执行。也就是这个任务进行排队，等正式工忙完了继续做。

ThreadFactory：线程工厂。生产线程，由你自己去定义你想生产什么样的线程。

RejectedExecutionHandler：拒绝策略。当你的最大线程与阻塞队列都满了。这个时候，你已经接收不了新的任务进行处理了。所以设置拒绝策略。相当于就是我所有的员工和临时工都在工作了，并且排队的任务都满了，应对这样的情况，你打算如何做。

除此之外还有一个重要的参数：

/*** If false (default), core threads stay alive even when idle.* If true, core threads use keepAliveTime to time out waiting* for work.*/privatevolatileboolean allowCoreThreadTimeOut;//是否允许核心线程数超时退出。

该参数有在特定的业务场景下有很大的意义。比如：你的业务只在晚上需要执行，其余时间无需执行。那么为何不把资源让出来，白天的时候，可以让其他业务占有这些资源去执行呢。

3.ThreadExecutorPool线程池重要源码解析

由该类图可知，Executor执行器定义执行方法，ExecutorService定义线程池操作的基本方法，AbstractExecutorService定义了线程池操作的方法模板。

ThreadPoolExecutor任务执行流程图

1.首先是构造方法

基本的参数校验与赋值，简单代码不过多赘述。

publicThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {基本的参数校验if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)  thrownewIllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)thrownewNullPointerException();this.acc = System.getSecurityManager() == null ?null :AccessController.getContext();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

2.线程执行的方法

public Future<?> submit(Runnable task) {if (task == null) thrownewNullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);//将线程对象封装成RunnableFutureexecute(ftask);//任务执行return ftask;}public <T> Future<T> submit(Callable<T> task) {if (task == null) thrownewNullPointerException();RunnableFuture<T> ftask = newTaskFor(task);//将线程对象封装成RunnableFutureexecute(ftask);//任务执行return ftask;}public <T> Future<T> submit(Runnable task, T result) {if (task == null) thrownewNullPointerException();RunnableFuture<T> ftask = newTaskFor(task, result);//将线程对象封装成RunnableFutureexecute(ftask);//任务执行return ftask;}

publicvoidexecute(Runnable command) {if (command == null)thrownewNullPointerException();intc= ctl.get();//获取当前的线程池状态。单个参数，保存了线程池的状态以及线程数量if (workerCountOf(c) < corePoolSize) { //当线程数量小于核心线程数if (addWorker(command, true)) //直接添加任务，运行线程return;c = ctl.get();}if (isRunning(c) && workQueue.offer(command)) {//如果核心线程数已经满了，那么直接添加到阻塞队列。intrecheck= ctl.get();if (! isRunning(recheck) && remove(command))//线程池不是running状态，执行拒绝策略。reject(command);elseif (workerCountOf(recheck) == 0)//线程池线程数量不能为0,需要有一个线程对线程池的后续操作进行处理，比如关闭线程池addWorker(null, false);}elseif (!addWorker(command, false))//当核心线程与阻塞队列都满了的时候，直接添加任务到非核心线程运行。添加失败直接执行拒绝策略reject(command);}

1.关于ctl.get()方法的解释---利用了单个变量，保存了线程池状态以及线程数量的值

privatefinalAtomicIntegerctl=newAtomicInteger(ctlOf(RUNNING, 0));privatestaticfinalintCOUNT_BITS= Integer.SIZE - 3;privatestaticfinalintCAPACITY= (1 << COUNT_BITS) - 1;// runState is stored in the high-order bitsprivatestaticfinalintRUNNING= -1 << COUNT_BITS; //运行状态 正常执行任务privatestaticfinalintSHUTDOWN=0 << COUNT_BITS; //关闭线程池，不再接收新任务privatestaticfinalintSTOP=1 << COUNT_BITS; //关闭线程池，所有任务停止privatestaticfinalintTIDYING=2 << COUNT_BITS; //中间状态privatestaticfinalintTERMINATED=3 << COUNT_BITS; //线程池已经关闭// Packing and unpacking ctlprivatestaticintrunStateOf(int c)     { return c & ~CAPACITY; }privatestaticintworkerCountOf(int c)  { return c & CAPACITY; }privatestaticintctlOf(int rs, int wc) { return rs | wc; }

2.addWorker方法

privatebooleanaddWorker(Runnable firstTask, boolean core) {retry:for (;;) {intc= ctl.get();//获取ctl的快照保存在栈上intrs= runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&   //如果线程池已经关闭，或者（当前线程池关闭状态当前任务是空且当前工作队列不为空）不满足的情况下直接返回! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))returnfalse;for (;;) {intwc= workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))returnfalse;if (compareAndIncrementWorkerCount(c))//CAS修改线程池ctl变量，增加线程数break retry; //添加成功直接退出c = ctl.get();  // 添加不成功，为了保证多线程运行的安全性，重新获取if (runStateOf(c) != rs)//当前线程池状态发生改变continue retry; //直接重新运行retry循环体// else CAS failed due to workerCount change; retry inner loop}}booleanworkerStarted=false;booleanworkerAdded=false;Workerw=null;try {w = newWorker(firstTask); //生成自定义的线程wokerfinalThreadt= w.thread;if (t != null) {finalReentrantLockmainLock=this.mainLock;//这个代码没有意义，mainLock定义的变量为final。可以直接使用mainLock.lock();//添加work使用锁，保证添加任务的原子性。try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.intrs= runStateOf(ctl.get());if (rs < SHUTDOWN || //线程池处于running状态(rs == SHUTDOWN && firstTask == null)) {//线程池处于showdown状态但是firstTask为空。if (t.isAlive()) // precheck that t is startablethrownewIllegalThreadStateException();workers.add(w);ints= workers.size();if (s > largestPoolSize)//保存当前线程池中线程的最大数量largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {//添加成功，运行线程t.start();workerStarted = true;}}} finally {if (! workerStarted)//线程启动失败addWorkerFailed(w);//移除work，减少线程数量}return workerStarted;}

t.start()执行线程任务

//Worker类中实际执行任务的方法 publicvoidrun() {runWorker(this);}
finalvoidrunWorker(Worker w) {Threadwt= Thread.currentThread();Runnabletask= w.firstTask;w.firstTask = null;w.unlock(); // allow interrupts  //将原始的线程状态为-1修改为0,后续通过getState()>=0获取线程是否已经运行的状态，允许线程中断。-1默认为初始化，此处需要进行处理booleancompletedAbruptly=true;try {while (task != null || (task = getTask()) != null) {//task不等于空直接运行，task等于空从workerQueue阻塞队列获取任务w.lock();// If pool is stopping, ensure thread is interrupted;// if not, ensure thread is not interrupted.  This// requires a recheck in second case to deal with// shutdownNow race while clearing interruptif ((runStateAtLeast(ctl.get(), STOP) ||//线程池运行状态大于等于STOP(Thread.interrupted() && //线程是否已经被中断了runStateAtLeast(ctl.get(), STOP))) &&//鲜橙汁运行状态大于等于STOP!wt.isInterrupted())//判断任务的线程如果没有被中断wt.interrupt();//中断当前任务线程try {beforeExecute(wt, task);//钩子函数，实际任务运行之前做处理Throwablethrown=null;try {task.run();//执行实际任务代码} catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {thrown = x; thrownewError(x);} finally {afterExecute(task, thrown);//钩子函数，实际任务运行之后做处理}} finally {task = null;//将任务置空w.completedTasks++;//任务完成数加1w.unlock();}}completedAbruptly = false;//执行过程中是否发成异常} finally {processWorkerExit(w, completedAbruptly);}}

//执行任务退出操作privatevoidprocessWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // 如果有异常中断导致任务结束decrementWorkerCount();//将线程数量减1finalReentrantLockmainLock=this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;//完成的任务数量累加workers.remove(w);//从workers的任务集合中移除当前任务} finally {mainLock.unlock();}tryTerminate();//尝试关闭线程池intc= ctl.get();//获取当前线程池的最新状态if (runStateLessThan(c, STOP)) {//如果当前任务状态小于STOPif (!completedAbruptly) {//当前任务执行无异常发生intmin= allowCoreThreadTimeOut ? 0 : corePoolSize;//根据allowCoreThreadTimeOut参数获取最小的线程数量if (min == 0 && ! workQueue.isEmpty())//如果核心线程允许退出，并且工作队列不为空min = 1;//设置最小值为1，因为最后需要有线程去执行线程池的后续处理，所有线程都没了，后续线程池退出无线程处理if (workerCountOf(c) >= min)//如果工作的线程数量大于等最小值return; // replacement not needed  直接返回}addWorker(null, false);//如果当前线程数已经小于最小线程数，那么需要保证最小线程数在运行，所以需要有保证线程池的正常运行，添加一个空任务。}}

private Runnable getTask() {booleantimedOut=false; // Did the last poll() time out?for (;;) {intc= ctl.get();//获取当前线程池状态intrs= runStateOf(c);//获取当前运行状态// Check if queue empty only if necessary.if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {//如果线程池状态大于等于SHUTDOWN并且(线程数量大于等于STOP或者工作队列为空)decrementWorkerCount();//将线程池中线程数量减1returnnull;}intwc= workerCountOf(c);//获取当前线程池的线程数量// Are workers subject to culling?booleantimed= allowCoreThreadTimeOut || wc > corePoolSize;//判断是否运行核心线程数超时，判断是否需要超时机制if ((wc > maximumPoolSize || (timed && timedOut))//工作线程大于最大线程池数量或者允许超时并且有超时的情况&& (wc > 1 || workQueue.isEmpty())) {//并且线程池线程数量大于1或者阻塞队列为空if (compareAndDecrementWorkerCount(c))//CAS操作将线程池数量减1returnnull;//返回空continue;//CAS失败继续}try {Runnabler= timed ?//允许超时从队列中拿任务并等待keepAliveTime时间workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();阻塞等待if (r != null)//获取的任务不为空return r;//直接返回timedOut = true;//如果为空，超时标志位为true} catch (InterruptedException retry) {timedOut = false;}}}

3.addWorkerFailed方法解析

privatevoidaddWorkerFailed(Worker w) {finalReentrantLockmainLock=this.mainLock;mainLock.lock();//获取锁try {if (w != null)//work不是空workers.remove(w);//直接从workers中移除当前任务decrementWorkerCount();//加个ctl中的woker数量减少tryTerminate();//如果线程池已经是showdown状态，尝试让线程池停止。多线程协作的函数} finally {mainLock.unlock();}}

3.线程池关闭shutdown方法

publicvoidshutdown() {finalReentrantLockmainLock=this.mainLock;mainLock.lock();try {checkShutdownAccess();//检查关闭权限，可以忽略advanceRunState(SHUTDOWN);//线程池状态递进，由running变为shutdowninterruptIdleWorkers();//中断所有空闲线程onShutdown(); // hook for ScheduledThreadPoolExecutor钩子函数，调度线程池使用} finally {mainLock.unlock();}tryTerminate();//尝试将线程池关闭。}

1.advanceRunState方法解析

privatevoidadvanceRunState(int targetState) {for (;;) {intc= ctl.get();//获取当前的线程状态if (runStateAtLeast(c, targetState) ||//当前状态已经是大于等于shutdown直接退出ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))//cas操作将线程状态改为targetState。break;}}

2.interruptIdleWorkers方法解析

privatevoidinterruptIdleWorkers() {interruptIdleWorkers(false);}privatevoidinterruptIdleWorkers(boolean onlyOne) {finalReentrantLockmainLock=this.mainLock;mainLock.lock();//获取锁try {for (Worker w : workers) {//遍历works中所有的工作任务Threadt= w.thread;if (!t.isInterrupted() && w.tryLock()) {//如果没有被中断过，并且可以获得锁，证明属于空闲线程try {t.interrupt();//将线程中断，打上中断标志位} catch (SecurityException ignore) {} finally {w.unlock();//解锁}}if (onlyOne)//只中断一个线程标识break;}} finally {mainLock.unlock();}}

4.shutdownNow方法解析

public List<Runnable> shutdownNow() {List<Runnable> tasks;finalReentrantLockmainLock=this.mainLock;mainLock.lock();try {checkShutdownAccess();//权限检查advanceRunState(STOP);//状态递进 详细方法见上面interruptWorkers();//中断所有启动的work线程tasks = drainQueue();//将所有未执行的任务出队保存} finally {mainLock.unlock();}tryTerminate();//尝试关闭线程池return tasks;}

1.interruptWorkers方法解析

privatevoidinterruptWorkers() {finalReentrantLockmainLock=this.mainLock;mainLock.lock();//获取锁try {for (Worker w : workers)//遍历所有woker进行处理w.interruptIfStarted();} finally {mainLock.unlock();}}voidinterruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {//当前work的状态大于0并且线程不为空且线程未被中断try {t.interrupt();} catch (SecurityException ignore) {}}}使用getState() >= 0表示当前线程已经启动，runWorker方法中会将其状态从-1改变。证明线程已经启动Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}

2.drainQueue方法解析

//标准的入队和出队功能不做过多注释   private List<Runnable> drainQueue() {BlockingQueue<Runnable> q = workQueue;ArrayList<Runnable> taskList = newArrayList<Runnable>();q.drainTo(taskList);if (!q.isEmpty()) {for (Runnable r : q.toArray(newRunnable[0])) {if (q.remove(r))taskList.add(r);}}return taskList;}

5.tryTerminate方法解析

finalvoidtryTerminate() {for (;;) {intc= ctl.get();//获取当前线程状态ctlif (isRunning(c) ||//线程池正在运行runStateAtLeast(c, TIDYING) ||//线程池状态大于等于TIDYING，有其他线程已经改变线程池状态为TIDYING或者TERMINATED了(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))//线程池状态等于shutdown并且工作队列不为空。return;//以上三种情况线程池无法关闭，需要继续处理if (workerCountOf(c) != 0) { // Eligible to terminate//当前工作线程数量不等于0interruptIdleWorkers(ONLY_ONE);//中断线程且只中断一个return;}finalReentrantLockmainLock=this.mainLock;mainLock.lock();try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {//cas操作将线程池状态置为TIDYINGtry {terminated();//线程池终止} finally {ctl.set(ctlOf(TERMINATED, 0));//设置线程池状态为TERMINATEDtermination.signalAll();//信号唤醒所有等待线程}return;}} finally {mainLock.unlock();}// else retry on failed CAS}
}

4.总结

线程池的运用在项目中已经成为一种常态，作为一个开发人员最重要的了解其背后的设计原理以及流程，更好地运用线程池，方便提升项目程序的性能以及排查错误。在阅读对应的线程池源码时，我们只局限于单线程的思维，更多的是要去考虑当多线程并发执行时的临界条件。了解设计者的设计初衷、以及设计意图，能让你更好地在项目中运用并设计符合自己项目的线程池。以上是我个人对于线程池ThreadPoolExecutor的理解，不足之处，请多多指教。