零、概述
线程池是什么
在并发环境，系统有多少任务需要执行，多少资源需要投入，都充满不确定性，容易导致如下问题：
- 频繁申请/销毁资源和调度资源，将带来额外的消耗，可能会非常巨大。
- 对资源无限申请缺少抑制手段，易引发系统资源耗尽的风险。
- 系统无法合理管理内部的资源分布，会降低系统的稳定性。
- 线程过多会带来额外的开销（创建销毁线程的开销、调度线程的开销等），降低了计算机的整体性能。
线程池（Thread Pool）是一种基于池化思想关联线程的工具，经常出现在多线程服务器中，如MySQL，采用池化（Pooling）的思想：为了最大化收益并最小化风险，将资源统一在一起管理的思想。
计算机领域中，也存在其他池化的用法：
- 内存池(Memory Pooling)：预先申请内存，提升申请内存速度，减少内存碎片。
- 连接池(Connection Pooling)：预先申请数据库连接，提升申请连接的速度，降低系统的开销。
- 实例池(Object Pooling)：循环使用对象，减少资源在初始化和释放时的昂贵损耗。
线程池维护多个线程，等待分配可执行的任务。避免了一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分利用，主要有如下优点：
- 降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
- 提高响应速度：任务到达时，无需等待线程创建即可立即执行。
- 提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。
- 提供更多更强大的功能：线程池具备可拓展性，允许开发人员向其中增加更多的功能。比如延时定时线程池ScheduledThreadPoolExecutor，就允许任务延期执行或定期执行。
Java线程池框架
Executor框架是一个根据一组执行策略调用、调度、控制的异步任务框架，目的是提供一种将 任务提交 与 任务运行 分离的机制。主要有三类接口：
- Executor：一个运行新任务的简单接口；
- ExecutorService：扩展了Executor接口。添加了一些用来管理执行器生命周期和任务生命周期的方法；
- ScheduledExecutorService：扩展了ExecutorService。支持Future和定期执行任务。
一、ThreadPoolExecutor
执行流程
构造函数
public ThreadPoolExecutor(int corePoolSize, // 核心线程数int maximumPoolSize, // 非核心线程数long keepAliveTime,// 时间TimeUnit unit, // 时间单位BlockingQueue<Runnable> workQueue, // 队列ThreadFactory threadFactory, // 线程工厂RejectedExecutionHandler handler) // 拒绝策略
1、corePoolSize：核心线程池大小。参数是否生效取决于allowCoreThreadTimeOut的值，默认为false，即对核心线程没有超时限制，这种情况corePoolSize生效；如果为True，则核心线程允许超时，并且超时时间有keepAliveTime、unit决定。
2、maximumPoolSize：线程池中最大的存活线程数。对于超出corePoolSize部分的线程，无论allowCoreThreadTimeOut为false或true，都会超时。
3、keepAliveTime：线程池维护线程所允许的空闲时间。当线程池中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会等待，直到等待的时间超过了keepAliveTime；
TimeUnit是一个枚举类型，其包括：
NANOSECONDS ： 1微毫秒 = 1微秒 / 1000
MICROSECONDS ： 1微秒 = 1毫秒 / 1000
MILLISECONDS ： 1毫秒 = 1秒 /1000
SECONDS ： 秒
MINUTES ： 分
HOURS ： 小时
DAYS ： 天
4、workQueue：任务队列，维护着等待执行的Runnable对象。调用execute或submit方法时，如果线程池中没有可用的线程，则Runnable对象会放到该队列。
SynchronousQueue: 不存储元素的阻塞队列，队列接收到任务时，直接交给线程处理。使用时设置maximumPoolSize为Integer.MAX_VALUE，即无限大 LinkedBlockingQueue: 使用链表实现的有界阻塞队列，FIFO原则。 ArrayBlockingQueue: 使用数组实现的有界阻塞队列，FIFO原则。 DelayQueue: 实现PriorityBlockingQueue的延迟获取的无解队列。
5、threadFactory：创建线程的方式。
默认调用：Executors.defaultThreadFactory()，使用默认的ThreadFactory来创建线程时，会使新创建的线程具有相同的NORM_PRIORITY优先级并且是非守护线程，同时也设置了线程的名称。
6、handler
它是RejectedExecutionHandler类型的变量，表示线程池的饱和策略。如果阻塞队列满了并且没有空闲的线程，这时如果继续提交任务，就需要采取一种策略处理该任务。线程池提供了4种策略：
- AbortPolicy：直接抛出异常，这是默认策略；
- CallerRunsPolicy：用调用者所在的线程来执行任务；
- DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
- DiscardPolicy：直接丢弃任务；
二、常见的四种线程池
最终都是使用ThreadPoolExecutor来创建线程池。
newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);}
newCachedThreadPool
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>(),threadFactory);}
newSingleThreadScheduledExecutor
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));}
newScheduledThreadPool
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);}public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,new DelayedWorkQueue());}
三、源码分析
ctl-状态标志位
ctl-基本定义
ctt是对线程池的运行状态和线程池中有效线程的数量进行控制的字段，包含两个部分的主要信息：线程池的运行状态（runState）和 有效线程数（workerCount）。使用int类型保存，高3为保存runState，低29位保存workerCount。
使用Integer.toBinaryString()方法打印常量（static final)：
CAPACITY:11111111111111111111111111111 ~CAPACITY:11100000000000000000000000000000 RUNNING:11100000000000000000000000000000 SHUTDOWN:0 STOP:100000000000000000000000000000 TIDYING:1000000000000000000000000000000 TERMINATED:1100000000000000000000000000000
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// Integer.SIZE 是一个 static final 的int常量，值为 32private static final int COUNT_BITS = Integer.SIZE - 3;// 1左移29位减去1，表示的容量的最大值private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 运行状态标志位,参考上面的打印信息private static final int RUNNING = -1 << COUNT_BITS;private static final int SHUTDOWN = 0 << COUNT_BITS;private static final int STOP = 1 << COUNT_BITS;private static final int TIDYING = 2 << COUNT_BITS;private static final int TERMINATED = 3 << COUNT_BITS;
线程池一共有5个运行状态：
1、RUNNING：能接受新提交的任务，并且也能处理阻塞队列中的任务 2、SHUTDOWN：关闭状态，不能再接受新提交的任务，但可以继续处理阻塞队列中已保存的任务。线程池处于RUNNING状态时，调用shutdown方法进入该状态 3、STOP：不能接受新任务，也不能处理队列中的任务，会中断正在处理任务的线程。
- 在线程池处于 RUNNING 或 SHUTDOWN 状态时，调用 shutdownNow() 方法会使线程池进入到该状态；
- TIDYING：如果所有的任务都已终止了，workerCount (有效线程数) 为0，线程池进入该状态后会调用 terminated() 方法进入TERMINATED 状态。
- TERMINATED：在terminated() 方法执行完后进入该状态，默认terminated()方法中什么也没有做。
进入TERMINATED的条件如下：
- 线程池不是RUNNING状态；
- 线程池状态不是TIDYING状态或TERMINATED状态；
- 如果线程池状态是SHUTDOWN并且workerQueue为空；
- workerCount为0；
- 设置TIDYING状态成功。
与ctl有关计算方法
// 获取运行状态private static int runStateOf(int c) { return c & ~CAPACITY; }// 获取worker数量private static int workerCountOf(int c) { return c & CAPACITY; }//获取运行状态和活动线程数的值private static int ctlOf(int rs, int wc) { return rs | wc; }private static boolean runStateLessThan(int c, int s) {return c < s;}private static boolean runStateAtLeast(int c, int s) {return c >= s;}private static boolean isRunning(int c) {return c < SHUTDOWN;}
submit-提交需要执行的任务
这个方法是在AbstractExecutorService类中实现的
public <T> Future<T> submit(Callable<T> task) {if (task == null) throw new NullPointerException();// newTaskFor方法的作用是，创建一个RunnableFuture对象RunnableFuture<T> ftask = newTaskFor(task);// 执行逻辑execute(ftask);return ftask;}
execute-执行主逻辑
public void execute(Runnable command) {if (command == null)throw new NullPointerException();// AtomicInteger 的get方法，返回ctl值int c = ctl.get();// 如果工作线程数小于 核心线程数，则增加workerif (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}//如果线程池处于 RUNNING状态，尝试任务加入队列。// offer方法会判断队列容量是能够增加新任务，不能则返回false。if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 再次判断线程池状态if (! isRunning(recheck) && remove(command))reject(command);//如果线程池中有效线程数是0，则执行addworker。注意这里只是在线程池中增加线程，但不启动else if (workerCountOf(recheck) == 0)addWorker(null, false);}/*执行到此处，有两种情况:1、线程池不是RUNNING状态2、线程池是RUNNING状态，但是workerCount>=corePoolSize 且 workQueue已经满了此时，调用addWoker方法，第二参数是false，设置 线程上线未 maximumPoolSize，如果设置失败，则拒绝该任务*/else if (!addWorker(command, false))reject(command);}
- 首先检测线程池运行状态，如果不是RUNNING，则直接拒绝，线程池要保证在RUNNING的状态下执行任务。
- 如果workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务。
- 如果workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中。
- 如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。
- 如果workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。
addWorker方法
for(;;) 等效于 while(true)
private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// 三个条件：1、线程不为SHUTDOWN状态 2、任务不为空 3、队列不为空。if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);// 1、有效线程数超过容量 ，返回false// 2、core标志位。if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// 尝试增加workerCount，如果成功。则跳出第一个for循环if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctl// 如果现在状态不等于rs，说明状态已被改变，返回第一个for循环，继续执行。if (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);// 每个work对象创建一个线程final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {int rs = runStateOf(ctl.get());// 再次判断是否可以增加线程if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {// 执行t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}
t.start()方法，启动线程的时候，会调用Woker类中的run方法（worker实现了Runnable接口，也是一个线程）。
Worker类
private final class Workerextends AbstractQueuedSynchronizerimplements Runnable{/*** This class will never be serialized, but we provide a* serialVersionUID to suppress a javac warning.*/private static final long serialVersionUID = 6138294804551838833L;/** Thread this worker is running in. Null if factory fails. */final Thread thread;/** Initial task to run. Possibly null. */Runnable firstTask;/** Per-thread task counter */volatile long completedTasks;/*** Creates with given first task and thread from ThreadFactory.* @param firstTask the first task (null if none)*/Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}/** Delegates main run loop to outer runWorker */public void run() {runWorker(this);}// Lock methods//// The value 0 represents the unlocked state.// The value 1 represents the locked state.protected boolean isHeldExclusively() {return getState() != 0;}protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}protected boolean tryRelease(int unused) {setExclusiveOwnerThread(null);setState(0);return true;}public void lock() { acquire(1); }public boolean tryLock() { return tryAcquire(1); }public void unlock() { release(1); }public boolean isLocked() { return isHeldExclusively(); }void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}}
runWoker方法
注意task.run()方法可以看成是函数调用，需要执行完后才会执行后续的逻辑。
final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {while (task != null || (task = getTask()) != null) {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) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {beforeExecute(wt, task);Throwable thrown = null;try {task.run();} catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {thrown = x; throw new Error(x);} finally {afterExecute(task, thrown);}} finally {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}}
getTask方法
private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}
processWokerExit
private void private void processWorkerExit(Worker w, boolean completedAbruptly) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusteddecrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}tryTerminate();int c = ctl.get();if (runStateLessThan(c, STOP)) {if (!completedAbruptly) {int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;if (workerCountOf(c) >= min)return; // replacement not needed}addWorker(null, false);}}(Worker w, boolean completedAbruptly) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusteddecrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}tryTerminate();int c = ctl.get();if (runStateLessThan(c, STOP)) {if (!completedAbruptly) {int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;if (workerCountOf(c) >= min)return; // replacement not needed}addWorker(null, false);}}
tryTerminate
final void tryTerminate() {for (;;) {int c = ctl.get();if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))return;if (workerCountOf(c) != 0) { // Eligible to terminateinterruptIdleWorkers(ONLY_ONE);return;}final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {try {terminated();// 啥也没干} finally {ctl.set(ctlOf(TERMINATED, 0));termination.signalAll();}return;}} finally {mainLock.unlock();}// else retry on failed CAS}}
showdown方法
shutdown把线程池设置成SHUTDOWN状态，不支持加入新的任务。对阻塞的任务、在运行中的任务没有影响。
public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();advanceRunState(SHUTDOWN);// 中断空闲线程interruptIdleWorkers();// ThreadPoolExecutor中，该方法do nothingonShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();}tryTerminate();}
private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;// w.tryLock()锁判断，正在执行的worker返回false（锁是在runWoker的时候增加的）if (!t.isInterrupted() && w.tryLock()) {try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}}
showdownNow方法
对比 interruptWokers 和 interruptIdleWorkers 。
后者检测到Woker是运行时（加锁），则不执行interrupt，所以被阻塞的线程不会抛异常。
前者只要检测到线程中断标志位=false时，就会调用 interrupt方法。
因此，showdownNow会停止正在运行的Woker，这里的运行是指 Woker内的线程是阻塞状态的种。
public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();advanceRunState(STOP);// 中断所有线程interruptWorkers();// 取出没有完成的任务tasks = drainQueue();} finally {mainLock.unlock();}tryTerminate();return tasks;}
private void interruptWorkers() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers)w.interruptIfStarted();} finally {mainLock.unlock();}}void interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {t.interrupt();} catch (SecurityException ignore) {}}}
参考：
1、Java线程池实现原理及其在美团业务中的实践
2、深入理解Java线程池：ThreadPoolExecutor
3、java线程池工作原理和实现原理
4、TheadPoolExecutor 中的Worker为什么要加锁?
若有收获，就点个赞吧
0 人点赞
