ThreadPoolExecutor源码阅读

public class ThreadPoolExecutor extends AbstractExecutorService { //线程池的运行状态（整形的高3位）和运行的worker数量（低29位） private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; private static final int CAPACITY = (1 << COUNT_BITS) - 1; /** * RUNNING 可接受新任务，可处理已经进入阻塞队列的任务 * SHUTDOWN 不接受新任务，可处理已经进入阻塞队列的任务 * STOP 不接受新任务，不处理已经进入阻塞队列的任务并且中断正在运行的任务 * TIDYING 所有的任务都已经终止，workerCount为0， 线程转化为TIDYING状态并且调用terminated钩子函数 * TERMINATED terminated钩子函数已经运行完成 */ 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; private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; } //阻塞队列 private final BlockingQueue<Runnable> workQueue; //可重入锁 private final ReentrantLock mainLock = new ReentrantLock(); 存放工作线程集合 private final HashSet<Worker> workers = new HashSet<Worker>(); // 终止条件 private final Condition termination = mainLock.newCondition(); // 最大线程池容量 private int largestPoolSize; // 已完成任务数量 private long completedTaskCount; //线程工厂 private volatile ThreadFactory threadFactory; //拒绝执行处理器 private volatile RejectedExecutionHandler handler; // 线程池中超过corePoolSize数目的空闲线程最大存活时间 private volatile long keepAliveTime; // 是否允许核心线程超时 private volatile boolean allowCoreThreadTimeOut; // 核心池的大小 private volatile int corePoolSize; // 最大线程池大小 private volatile int maximumPoolSize; // 默认拒绝执行处理器 private static final RejectedExecutionHandler defaultHandler = new AbortPolicy(); private final class Worker extends AbstractQueuedSynchronizer implements Runnable { private static final long serialVersionUID = 6138294804551838833L; //worker所对应的线程 final Thread thread; //worker的执行任务 Runnable firstTask; //已完成任务数量 volatile long completedTasks; Worker(Runnable firstTask) { setState(-1); this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); } 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(); } } //尝试终止线程池 final void tryTerminate() { for (;;) { int c = ctl.get(); if (isRunning(c) || //运行状态 runStateAtLeast(c, TIDYING) || //TIDYIN或者TERMINATED状态 (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) //SHUTDOWN状态且阻塞任务队列非空 return;// 不能终止，直接返回 if (workerCountOf(c) != 0) { // Eligible to terminate interruptIdleWorkers(ONLY_ONE);// 仅仅中断一个空闲的worker return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {// 比较并设置线程池控制状态为TIDYING try { terminated(); } finally { ctl.set(ctlOf(TERMINATED, 0));// 设置线程池控制状态为TERMINATED termination.signalAll();// 释放在termination条件上等待的所有线程 } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } } private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) {// 线程未被中断并且成功获得锁 try { t.interrupt();// 中断线程 } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne)// 若只中断一个，则跳出循环 break; } } finally { mainLock.unlock(); } } private void interruptIdleWorkers() { interruptIdleWorkers(false); } private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get();//获取线程池控制状态 int rs = runStateOf(c);//线程池运行状态 // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c);// worker数量 if (wc >= CAPACITY || // worker数量大于等于最大容量 wc >= (core ? corePoolSize : maximumPoolSize))// worker数量大于等于核心线程池大小或者最大线程池大小 return false; if (compareAndIncrementWorkerCount(c))// 比较并增加worker的数量 break retry;//结束循环 c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs)// 此次的状态与上次获取的状态不相同 continue retry; } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask);//创建一个worker final Thread t = w.thread;// 获取worker对应的线程 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 startable throw new IllegalThreadStateException(); workers.add(w);// 将worker添加到worker集合 int s = workers.size();// 获取worker集合的大小 if (s > largestPoolSize) largestPoolSize = s; workerAdded = true;// 设置worker已被添加标识 } } finally { mainLock.unlock(); } if (workerAdded) { t.start();// 开始执行worker workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; } private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) workers.remove(w); decrementWorkerCount(); tryTerminate(); } finally { mainLock.unlock(); } } //worker退出 private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 将worker完成的任务添加到总的完成任务中 completedTaskCount += w.completedTasks; workers.remove(w);// 从workers集合中移除该worker } finally { mainLock.unlock(); } // 尝试终止线程池 tryTerminate(); int c = ctl.get(); if (runStateLessThan(c, STOP)) { if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty())// 允许核心线程超时且workQueue阻塞队列不为空 min = 1; if (workerCountOf(c) >= min) return; // replacement not needed } addWorker(null, false);// 添加worker } } //从workerQueue阻塞队列中获取Runnable对象 private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get();// 获取线程池控制状态 int rs = runStateOf(c);// 线程池运行状态 // 大于等于SHUTDOWN（表示调用了shutDown）并且（大于等于STOP（调用了shutDownNow）或者worker阻塞队列为空） if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount();// 减少worker的数量 return null;// 返回null，不执行任务 } int wc = workerCountOf(c);// 获取worker数量 // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null;// 返回null，不执行任务，该worker会退出 continue; } try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();// 一直等待，直到有任务 if (r != null) return r;//返回任务 timedOut = true;// 等待指定时间后，没有获取元素，则超时 } catch (InterruptedException retry) { timedOut = false; } } } final void runWorker(Worker w) { Thread wt = Thread.currentThread();//获取当前线程 Runnable task = w.firstTask; 获取w的firstTask w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true;//异常标志 try {// 任务不为null或者阻塞队列还存在任务 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 interrupt if ((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); } } 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.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; } /** * 进行下面三步 * * 1. 如果运行的线程小于corePoolSize,则尝试使用用户定义的Runnalbe对象创建一个新的线程 * 调用addWorker函数会原子性的检查runState和workCount，通过返回false来防止在不应 * 该添加线程时添加了线程. * 2. 如果一个任务能够成功入队列，在添加一个线程时仍需要进行双重检查（因为在前一次检查后 * 该线程死亡了），或者当进入到此方法时，线程池已经shutdown了，所以需要再次检查状态， * 若有必要，当停止时还需要回滚入队列操作，或者当线程池没有线程时需要创建一个新线程. * 3. 如果无法入队列，那么需要增加一个新线程，如果此操作失败，那么就意味着线程池已经shutdown * 或者已经饱和了，所以拒绝任务. */ public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get();// 获取线程池控制状态 if (workerCountOf(c) < corePoolSize) {// worker数量小于corePoolSize if (addWorker(command, true))// 添加worker,返回 return; c = ctl.get();// 不成功则再次获取线程池控制状态 } // 线程池处于RUNNING状态，将command添加进workQueue队列 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get();// 再次检查，获取线程池控制状态 // 线程池不处于RUNNING状态，将command从workQueue队列中移除 if (! isRunning(recheck) && remove(command)) reject(command);//拒绝执行command else if (workerCountOf(recheck) == 0)// worker数量等于0 addWorker(null, false); } else if (!addWorker(command, false))// 添加worker失败 reject(command); } public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(SHUTDOWN);// 设置线程池控制状态为SHUTDOWN interruptIdleWorkers();// 中断空闲worker onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } tryTerminate();// 尝试终止线程池 } }