线程池有哪些状态

1. RUNNING: 接收新的任务，且执行等待队列中的任务 Accept new tasks and process queued tasks
2. SHUTDOWN: 不接收新任务，但执行等待队列中的任务 Don't accept new tasks, but process queued tasks
3. STOP: 不接收新任务，且不执行等待队列中的任务，且中断正在执行的任务 Don't accept new tasks, don't process queued tasks,and interrupt in-progress tasks
4. TIDYING: 所有任务都终止了，工作线程数为0，并执行terminated()回调函数 All tasks have terminated, workerCount is zero,the thread transitioning to state TIDYING , will run the terminated() hook method
5. TERMINATED: terminated() 回调函数执行完毕 has completed

这些状态代表的数值是递增的。
状态变迁有如下几种情况。

1. RUNNING -> SHUTDOWN 用户手动调用shutdown()方法 On invocation of shutdown(), perhaps implicitly in finalize()
2. (RUNNING or SHUTDOWN) -> STOP 用户手动调用shutdownNow()方法 invocation of shutdownNow()
3. SHUTDOWN -> TIDYING 等待队列和池子本身都空了
4. STOP -> TIDYING 池子空了 When pool is empty
5. TIDYING -> TERMINATED terminated() 回调函数执行完毕 When the terminated() hook method has completed

awaitTermination方法会阻塞直到线程池的状态为terminated。

如何判断线程池当前的状态

ctl 是一个31位的二进制数，其中线程池状态（高3位），工作线程个数（低29位），且ThreadPoolExecutor内部提供了获取它们的方法，在读其源码的过程中可以经常看到。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));/** ThreadPoolExecutor内部用于获取线程池状态 **/
private static int runStateOf(int ctl)     { return ctl & ~CAPACITY; }/** ThreadPoolExecutor内部用于获取工作线程数 **/
private static int workerCountOf(int ctl)  { return ctl & CAPACITY; }/** ThreadPoolExecutor内部用于获取线程池是否正在运行 **/
private static boolean isRunning(int c) {return c < SHUTDOWN;
}

线程池里都有哪些线程

核心线程：核心线程池中的线程
非核心线程：任务队列满后，如小于线程池最大线程数，则创建非核心线程。非核心线程空闲后，其存活时间取决于用户设置的keepAliveTime，到期后回收。
工作线程：工作状态的线程，可以是核心线程，也可以是非核心线程。工作线程存放在名为workers的HashSet中。

 /*** Set containing all worker threads in pool. Accessed only when* holding mainLock.*/private final HashSet<Worker> workers = new HashSet<Worker>();

核心线程均忙，提交的任务存放在哪里（线程池中的队列）

这个队列用于存储核心线程池满后提交的任务，并等待核心线程空闲时，将保存的任务吐给核心线程（工作线程 worker threads）进行执行。

队列不能通过workQueue.poll() == null 来确认队列是否为空，因为诸如DelayQueue之类的队列是允许存放空值的。判空可以用workQueue.isEmpty()来进行。

  private final BlockingQueue<Runnable> workQueue;

核心流程

创建线程池（new ThreadPoolExecutor）

corePoolSize	核心线程数 1. 核心线程哪怕空闲也不会被杀掉，除非allowCoreThreadTimeOut设为true，则会在keepAliveTime结束后被杀死
maximunPoolSize	最大线程数
keeyAliveTime	非核心线程空闲后的最大存活时间，若allowCoreThreadTimeOut设为true，则也是核心线程空闲后的最大存活时间
TimeUnit	存活时间的时间单位
workQueue	阻塞队列 1. 基于数组的有界ArrayBlockingQueue 2. 基于链表的无界LinkedBlockingQueue 3. 最多只有一个元素的SynchronrousQueue 4. 优先级队列PriorityBlockingQueue
ThreadFactory	创建线程的工厂
RejectedExecutionHandler	队列满，而工作线程数=最大线程数时，再提交任务会触发的拒绝策略 1. 直接抛出异常AbortPolicy 2. 使用调用者所在线程来运行任务CallerRunsPolicy 3. 丢弃等待最久的任务DiscardOldestPolicy 4. 默默丢弃，不抛异常DiscardPolicy

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();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;}

提交任务（submit）

ThreadPoolExecutor#submit后主要执行execute() 方法，依然遵循线程池的灵魂三步。

创建核心线程
核心线程满则加入队列
队列满则创建非核心线程

首先不论向线程池提交的是Runnable任务还是Callable任务，都会被包装为RunnableFuture对象，并被当作firstTask。（就是下面源码中的firstTask）

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();1. 若当前工作线程数<核心线程数阈值，则新开启一个工作线程，且该工作线程为核心线程，执行任务可见非running状态是无法提交的任务，也应验了shutdown()方法可以阻止新任务的提交if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}2. 若核心线程满了，线程池也处于running状态，则向任务队列提交任务if (isRunning(c) && workQueue.offer(command)) {// 二次检查int recheck = ctl.get();// 如果当前线程池状态不是RUNNING则从队列删除任务，并执行拒绝策略if (! isRunning(recheck) && remove(command))reject(command);// 否者如果当前线程池线程空，则添加一个线程else if (workerCountOf(recheck) == 0)addWorker(null, false);}3. 如果队列满了，则新开启工作线程，且工作线程为非核心线程执行任务，新增失败则执行拒绝策略else if (!addWorker(command, false))reject(command);
}新开启工作线程方法
private boolean addWorker(Runnable firstTask, boolean core) {retry:1. 开始自旋for (;;) {int c = ctl.get();int rs = runStateOf(c);2. 可见非running状态是无法提交任务的，也应验了shutdown()方法可以阻止新任务的提交if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {3. 如果要添加的工作线程是核心线程，但核心线程满了，则快速失败如果要添加的工作线程非核心线程，但已经到达最大线程数，同样快速失败int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;4. 工作线程+1，成功后跳出自旋if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}5. 成功跳出自旋后...boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {6. 为当前任务创建工作线程w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {7. 由于要操作workers这个HashSet了，上把锁，然后进入临界区final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 二次检查int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {8. 若当前线程已经是工作中线程了，快速失败if (t.isAlive())throw new IllegalThreadStateException();9. 向workers中新增该工作线程workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;10. 标记工作线程新增成功workerAdded = true;}} finally {11. 解锁，退出临界区mainLock.unlock();}12. 此处可能无法立即执行，因为可能切换到另外一个线程去了，但回来后仍会根据是否成功添加工作线程，给予该工作线程start的机会if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;
}

再看创建的worker。

创建工作线程（new Worker）

可以发现worker实现了Runnable接口，又继承了AQS，可见worker本身可以是工作线程，又具备AQS协同线程和管理队列的能力。（AbstractQueueSynchronizer实现的，如果对AQS不熟悉，请先跳去看AQS)。

所以Worker不仅继承了AQS中waitQueue，conditionQueue等概念，还重写了AQS提供的几个抽象API。包括：

tryAquire：若state为0，则从0置为1，并且一定会让当前线程获取独占锁（因为若compareAndSetState结束后切换执行另外的线程，另外的线程compareAndSetState必然返回false）。若state非0，则置1失败，表示线程无法独占式获取锁。
tryRelease：将state恢复为0，当前线程释放独占锁。

由此可见，在ThreadPoolExecutor中，state要么为1，表示当前线程独占式获取锁。要么为0，表示当前线程释放了独占锁。要么为-1，表示初始状态。

isHeldExclusively：判断当前线程是否获取独占锁，state=1表示有，state=0表示无。

而其中又调用了runWorker方法。

 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) {}}}}

启动工作线程（run worker）

进入runWorker后，首先调用的是worker.unlock()，这是在利用其继承自AQS的能力，跟踪代码可以发现，其实就是将state从-1置0，将独占线程的标记置空，即恢复初始状态。
然后进入while循环中，这样这个工作线程算是正式开始工作了，不断从wait队列中获取任务执行。
当前工作线程尝试获取独占锁。（基于AQS的aquire，获取失败就中断当前线程并加入waitQueue）
然后判断下线程池是不是要关了，要关了就中断当前线程。
在执行task.run之前，先执行beforeExecute，它可能会抛出异常，使得当前的线程死亡（线程死亡，而非任务中断）。
执行task.run。
执行task.run之后，会执行afterExecute，它会把当前的任务和其抛出的异常打包处理，注意只有Runnable才能将未检查异常抛出（Thrown不为空），而Callable会将异常封装到future对象中（Thrown为空）。
释放独占锁。
while循环都结束了，说明这条线程坚持到最后都没因为run中用户业务代码的异常导致线程死亡，completedAbruptly=false表示没有因业务报错线程快速死亡，这是条幸运的线程
所有任务都执行完后，调用processWorkerExit，尝试将线程池终止，

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {2. 然后进入while循环中，这样这个工作线程算是正式开始工作了，不断从wait队列中获取任务执行。while (task != null || (task = getTask()) != null) {3. 每获取一个任务，当前工作线程就获取独占锁。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 {6. 执行task.runtask.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);}}

工作线程获取锁成功

"启动工作线程"章节中的第三步，worker.lock()。
调用AQS的aquire()。
AQS中，调用被Worker重写的tryAquire方法，回到worker中，若当前state=0（没有线程获取独占锁），则将state置1，并标记当前线程获取独占锁，返回成功。

 1. ThreadPoolExecutor的Worker类中，调用了AQS的aquire方法public void lock()        { acquire(1); }2. 调用AQS的aquire()。public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 5. 再调用acquireQueued方法selfInterrupt();}3. AQS中，调用被Worker重写的tryAquire方法，回到worker中，若当前state=0（没有线程获取独占锁），则将state置1，并标记当前线程获取独占锁，返回成功。protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}

工作线程获取锁失败

"启动工作线程"章节中的第三步，worker.lock()。
调用AQS的aquire()。
AQS中，调用被Worker重写的tryAquire方法，回到worker中，若当前state=0（没有线程获取独占锁），则将state置1，并标记当前线程获取独占锁，返回成功。
若tryAquire返回失败（其他线程已获取独占锁），则用addWaiter将当前线程设为独占式节点插入waitQueue尾部。
再调用acquireQueued方法：
1. 开启自旋
2. 若发现头指针指向当前节点，说明虽然之前获取锁失败，但是waitQueue竟然不知不觉空了，当前节点又得到了尝试获取锁的机会。
3. 若当前节点乖乖排在最后，则同时传入当前节点和其前驱，调用shouldParkAfterFailAquired方法，判断是否需要将当前线程挂起（park）。
  1. 若前驱节点的状态为signal，表明当前节点的前驱离开waitQueue后，会调用unpark唤醒当前节点。所以当前节点需要被挂起park，才能等待unpark唤醒（signal的含义），则shouldParkAfterFailAquired返回true。
  2. 若前驱的状态为cancel（只有cancel>0），则表示该前驱废弃，往前继续回溯，直到找到非取消状态的节点，重新作为当前节点的前驱，返回false，当前节点无需挂起。
  3. 若前驱为其他状态，如0、condition、propergate，则将前驱的状态改为signal，然后返回false，表示当前节点无需挂起。
4. 若需要挂起，则使用parkAndCheckInterrupt，其中使用LockSupport.park()将当先线程挂起，静静等待其他线程将它唤醒unpark或者中断interrupt。
5. 如果陡然间，当前线程发现自己从沉睡中被唤醒，则检查自己的中断状态，判断是被正常叫醒，还是被中断，若是被中断则将interrupted标记为设为true。
6. 被唤醒后继续自旋，判断当前自己是不是位于waitQueue头部，是则尝试获取锁，失败了继续将自己挂起。
7. 如果是中断唤醒，interrupted为true，而在5.6中获取锁成功，则对不起，继续中断吧。
8. 最后如果失败了，调用cancelAcquire，暂时未分析。

 1. ThreadPoolExecutor的Worker类中，调用了AQS的aquire方法public void lock()        { acquire(1); }2. 调用AQS的aquire()。public final void acquire(int arg) {4. 若tryAquire返回失败（其他线程已获取独占锁），则用addWaiter将当前线程设为独占式节点插入waitQueue尾部。if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 5. 再调用acquireQueued方法selfInterrupt(); 5.7 如果是中断唤醒，interrupted为true，而在5.6中获取锁成功，则对不起，继续中断吧。}3. AQS中，调用被Worker重写的tryAquire方法，回到worker中，若当前state=0（没有线程获取独占锁），则将state置1，并标记当前线程获取独占锁，返回成功。protected boolean tryAcquire(int unused) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;5.1 开启自旋for (;;) {final Node p = node.predecessor();5.2 若发现头指针指向当前节点，说明虽然之前获取锁失败，但是waitQueue竟然不知不觉空了，当前节点又得到了尝试获取锁的机会if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return interrupted;}5.3 若当前节点乖乖排在最后，则同时传入当前节点和其前驱，调用shouldParkAfterFailAquired方法，判断是否需要将当前线程挂起（park）if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) 5.4 若需要挂起，则使用parkAndCheckInterrupt，其中使用LockSupport.park()将当先线程挂起，静静等待其他线程将它唤醒unpark或者中断interrupt，才将interrupted标记为设为true。interrupted = true;}} finally {6. 最后如果失败了，调用cancelAcquire，暂时未分析。if (failed)cancelAcquire(node);}}private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;if (ws == Node.SIGNAL)5.3.1 若前驱节点的状态为signal，则表明当前节点 a.需要被挂起  b.需要等待unpark唤醒（signal的含义），则shouldParkAfterFailAquired返回true。return true;if (ws > 0) {5.3.2 若前驱的状态为cancel（只有cancel>0），则表示该前驱废弃，往前继续回溯，直到找到非取消状态的节点，重新作为当前节点的前驱，返回false，当前节点无需挂起。do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {5.3.3 若前驱为其他状态，如0、condition、propergate，则将前驱的状态改为signal，然后返回false，表示当前节点无需挂起。compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}

工作线程获取锁成功，但又得等资源

工作线程意外身亡

一些碎片源码

tryTerminate

尝试终止整个线程池从，但是条件非常苛刻。

无法终止线程池的原因包括两种：线程池已经终止 或 线程池仍有任务需要尽快消费。

可以细分为以下情况：

当前线程池的状态是running运行中。
当前线程池已经终止，即状态为TIDYING或TERMINATING，则无需再次终止多此一举。
用户调用了shutdown方法，但是任务队列不为空。
当前仍然有工作线程存活。

以上重重关卡都通过，才能尝试终止，上锁，将当前线程池状态置TIDYING，唤醒调用了awaitTermination的线程，解锁。

/*** Transitions to TERMINATED state if either (SHUTDOWN and pool* and queue empty) or (STOP and pool empty).  If otherwise* eligible to terminate but workerCount is nonzero, interrupts an* idle worker to ensure that shutdown signals propagate. This* method must be called following any action that might make* termination possible -- reducing worker count or removing tasks* from the queue during shutdown. The method is non-private to* allow access from ScheduledThreadPoolExecutor.*/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}}

processWorkerExit

private void processWorkerExit(Worker w, boolean completedAbruptly) {1. 统计整个线程池完成的任务个数,并从工作集里面删除当前woker，需要上锁才能操作worker这个hashSetfinal ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}2. 尝试将当前线程池终止tryTerminate();3. 如果当前状态小于stop，应保证尽快将任务消费完所以如果线程意外死亡（completedAbruptly=true）或当前线程数已经小于核心个数，则增加工作线程。如果任务全搞定，则线程池状态会变成tidy，> stop，自然不会再创建新工作线程了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);}
}

关联博客

java线程池ThreadPoolExecutor类详解 https://blog.csdn.net/qq_26950567/article/details/117435378

ReentrantLock源码解析 https://blog.csdn.net/qq_26950567/article/details/117442244

AbstractQueueSynchronizer详解 https://blog.csdn.net/qq_26950567/article/details/117440770

java线程池ThreadPoolExecutor类详解相关推荐

Java线程池七大参数详解和配置
目录一.corePoolSize核心线程数二.maximunPoolSize最大线程数三.keepAliveTime空闲线程存活时间四.unit空闲线程存活时间的单位五.workQueue线 ...
并发编程五：java并发线程池底层原理详解和源码分析
文章目录 java并发线程池底层原理详解和源码分析线程和线程池性能对比 Executors创建的三种线程池分析自定义线程池分析线程池源码分析继承关系 ThreadPoolExecutor源码分 ...
线程池invokeAll方法详解
线程池invokeAll方法详解问题起源与抽象问题排查与猜测猜测一:invokeAll 在异步执行后会不会同步等待线程执行完毕获取最终结果猜测二:队列里面可能存在第一次调用 invokeAll ...
Java 线程池(ThreadPoolExecutor)原理分析与使用 – 码农网
线程池的详解 Java 线程池(ThreadPoolExecutor)原理分析与使用 – 码农网 http://www.codeceo.com/article/java-threadpool-exec ...
basicdatasourcefactory mysql_Java基础-DBCP连接池(BasicDataSource类)详解
Java基础-DBCP连接池(BasicDataSource类)详解作者:尹正杰版权声明:原创作品,谢绝转载!否则将追究法律责任. 实际开发中"获得连接"或"释放资源 ...
Java线程池ThreadPoolExecutor使用和分析(三) - 终止线程池原理
相关文章目录: Java线程池ThreadPoolExecutor使用和分析(一) Java线程池ThreadPoolExecutor使用和分析(二) - execute()原理 Java线程池Thr ...
Java线程池ThreadPoolExecutor使用和分析
Java线程池ThreadPoolExecutor使用和分析(一) Java线程池ThreadPoolExecutor使用和分析(二) Java线程池ThreadPoolExecutor使用和分析(三 ...
Java中的Runtime类详解
Java中的Runtime类详解 1.类注释 /**Every Java application has a single instance of class Runtime that allows ...
stringtokenizer java_基于Java中的StringTokenizer类详解(推荐)
StringTokenizer是字符串分隔解析类型,属于:Java.util包. 1.StringTokenizer的构造函数 StringTokenizer(String str):构造一个用来解析 ...

java线程池ThreadPoolExecutor类详解