从“设计思想”到“源码解读”，一步一步深入解析Java线程池[通俗易懂]

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler);}public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler);}public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);}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 ==  || threadFactory ==  || handler == ) throw new PointerException(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler;}

欢迎大家来到IT世界,在知识的湖畔探索吧!

欢迎大家来到IT世界,在知识的湖畔探索吧!/** * The default thread factory */static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix;
 DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != ) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; }
 public Thread newThread(Runnable r) { Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; }}

欢迎大家来到IT世界,在知识的湖畔探索吧!import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.ThreadPoolExecutor;import java.util.concurrent.TimeUnit;
public class MyTest { public static void main(String[] args) { // 创建线程池 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 5, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(5)); // 向线程池提交任务 for (int i = 0; i < threadPool.getCorePoolSize(); i++) { threadPool.execute(new Runnable() { @Override public void run() { for (int x = 0; x < 2; x++) { System.out.println(Thread.currentThread().getName() + ":" + x); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } } }); }
 // 关闭线程池 threadPool.shutdown(); // 设置线程池的状态为SHUTDOWN，然后中断所有没有正在执行任务的线程 // threadPool.shutdownNow(); // 设置线程池的状态为STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表，该方法要慎用，容易造成不可控的后果 }}

// 使用默认线程工厂public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());}// 需要自定义线程工厂public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory);}

// 1\. 创建线程池对象，设置核心线程和最大线程数为5ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);// 2\. 创建Runnable（任务）Runnable task =new Runnable(){ public void run() { System.out.println(Thread.currentThread().getName() + "--->运行"); }};// 3\. 向线程池提交任务fixedThreadPool.execute(task);

public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));}// 为节省篇幅，省略了自定义线程工厂方式的源码

// 1\. 创建单线程线程池ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();// 2\. 创建Runnable（任务）Runnable task = new Runnable(){ public void run() { System.out.println(Thread.currentThread().getName() + "--->运行"); }};// 3\. 向线程池提交任务singleThreadExecutor.execute(task);

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize);}
// 继承了 ThreadPoolExecutorpublic class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService { // 构造函数，省略了自定义线程工厂的构造函数 public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); }
 // 延时执行任务 public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { ... } // 定时执行任务 public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {...}}

// 1\. 创建定时线程池ExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);// 2\. 创建Runnable（任务）Runnable task = new Runnable(){ public void run() { System.out.println(Thread.currentThread().getName() + "--->运行"); }};// 3\. 向线程池提交任务scheduledThreadPool.schedule(task, 2, TimeUnit.SECONDS); // 延迟2s后执行任务scheduledThreadPool.scheduleAtFixedRate(task,50,2000,TimeUnit.MILLISECONDS);// 延迟50ms后、每隔2000ms执行任务

public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());}

// 1\. 创建缓存线程池ExecutorService cachedThreadPool = Executors.newCachedThreadPool();// 2\. 创建Runnable（任务）Runnable task = new Runnable(){ public void run() { System.out.println(Thread.currentThread().getName() + "--->运行"); }};// 3\. 向线程池提交任务cachedThreadPool.execute(task);

/** * 在将来的某个时候执行给定的任务。任务可以在新线程中执行，也可以在现有的池线程中执行。 * 如果由于此执行器已关闭或已达到其容量而无法提交任务以供执行，则由当前的{@code RejectedExecutionHandler}处理该任务。 *  * @param command the task to execute 待执行的任务命令 */public void execute(Runnable command) { if (command == ) throw new PointerException(); /* * Proceed in 3 steps: *  * 1\. 如果运行的线程少于corePoolSize，将尝试以给定的命令作为第一个任务启动新线程。 * * 2\. 如果一个任务可以成功排队，那么我们仍然需要仔细检查两点，其一，我们是否应该添加一个线程 * （因为自从上次检查至今，一些存在的线程已经死亡），其二，线程池状态此时已改变成非运行态。因此，我们重新检查状态，如果检查不通过，则移除已经入列的任务，如果检查通过且线程池线程数为0，则启动新线程。 *  * 3\. 如果无法将任务加入任务队列，则将线程池扩容到极限容量并尝试创建一个新线程，如果失败则拒绝任务。 */ int c = ctl.get();
 // 步骤1：判断线程池当前线程数是否小于线程池大小 if (workerCountOf(c) < corePoolSize) { // 增加一个工作线程并添加任务，成功则返回，否则进行步骤2 // true代表使用coreSize作为边界约束，否则使用maximumPoolSize if (addWorker(command, true)) return; c = ctl.get(); } // 步骤2：不满足workerCountOf(c) < corePoolSize或addWorker失败，进入步骤2 // 校验线程池是否是Running状态且任务是否成功放入workQueue（阻塞队列） if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 再次校验，如果线程池非Running且从任务队列中移除任务成功，则拒绝该任务 if (! isRunning(recheck) && remove(command)) reject(command); // 如果线程池工作线程数量为0，则新建一个空任务的线程 else if (workerCountOf(recheck) == 0) // 如果线程池不是Running状态，是加入不进去的 addWorker(, false); } // 步骤3：如果线程池不是Running状态或任务入列失败，尝试扩容maxPoolSize后再次addWorker，失败则拒绝任务 else if (!addWorker(command, false)) reject(command);}

private boolean addWorker(Runnable firstTask, boolean core) { // 外层循环：判断线程池状态 retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c);
 /**  * 1.线程池为非Running状态（Running状态则既可以新增核心线程也可以接受任务） * 2.线程为shutdown状态且firstTask为空且队列不为空 * 3.满足条件1且条件2不满足，则返回false * 4.条件2解读：线程池为shutdown状态时且任务队列不为空时，可以新增空任务的线程来处理队列中的任务 */ if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask ==  && ! workQueue.isEmpty())) return false;
 // 内层循环：线程池添加核心线程并返回是否添加成功的结果 for (;;) { int wc = workerCountOf(c); // 校验线程池已有线程数量是否超限： // 1.线程池最大上限CAPACITY  // 2.corePoolSize或maximumPoolSize（取决于入参core） if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))  return false; // 通过CAS操作使工作线程数+1，跳出外层循环 if (compareAndIncrementWorkerCount(c))  break retry; // 线程+1失败，重读ctl c = ctl.get(); // Re-read ctl // 如果此时线程池状态不再是running，则重新进行外层循环 if (runStateOf(c) != rs) continue retry; // 其他 CAS 失败是因为工作线程数量改变了，继续内层循环尝试CAS对线程数+1 // else CAS failed due to workerCount change; retry inner loop } }
 /** * 核心线程数量+1成功的后续操作：添加到工作线程集合，并启动工作线程 */ boolean workerStarted = false; boolean workerAdded = false; Worker w = ; try { final ReentrantLock mainLock = this.mainLock; w = new Worker(firstTask); final Thread t = w.thread; if (t != ) { // 下面代码需要加锁：线程池主锁 mainLock.lock();  try { // 持锁期间重新检查，线程工厂创建线程失败或获取锁之前关闭的情况发生时，退出 int c = ctl.get(); int rs = runStateOf(c);
 // 再次检验线程池是否是running状态或线程池shutdown但线程任务为空 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == )) { // 线程已经启动，则抛出非法线程状态异常 // 为什么会存在这种状态呢？未解决 if (t.isAlive()) // precheck that t is startable throw 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 { //线程启动失败，则进入addWorkerFailed if (! workerStarted)  addWorkerFailed(w); } return workerStarted;}

private final class Worker extends AbstractQueuedSynchronizer implements 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. if factory fails. */ final Thread thread; 
 /** Initial task to run. Possibly . */ Runnable firstTask;
 /** Per-thread task counter */ volatile long completedTasks;
 /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task ( if none) */ // 通过构造函数初始化， Worker(Runnable firstTask) { //设置AQS的同步状态 // state：锁状态，-1为初始值，0为unlock状态，1为lock状态 setState(-1); // inhibit interrupts until runWorker 在调用runWorker前，禁止中断
 this.firstTask = firstTask; // 线程工厂创建一个线程 this.thread = getThreadFactory().newThread(this);  }
 /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); //runWorker()是ThreadPoolExecutor的方法 }
 // Lock methods // The value 0 represents the unlocked state. 0代表“没被锁定”状态 // The value 1 represents the locked state. 1代表“锁定”状态
 protected boolean isHeldExclusively() { return getState() != 0; }
 /** * 尝试获取锁的方法 * 重写AQS的tryAcquire()，AQS本来就是让子类来实现的 */ protected boolean tryAcquire(int unused) { // 判断原值为0，且重置为1，所以state为-1时，锁无法获取。 // 每次都是0->1，保证了锁的不可重入性 if (compareAndSetState(0, 1)) { // 设置exclusiveOwnerThread=当前线程 setExclusiveOwnerThread(Thread.currentThread());  return true; } return false; }
 /** * 尝试释放锁 * 不是state-1，而是置为0 */ protected boolean tryRelease(int unused) { setExclusiveOwnerThread();  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(); }
 /** * 中断（如果运行） * shutdownNow时会循环对worker线程执行 * 且不需要获取worker锁，即使在worker运行时也可以中断 */ void interruptIfStarted() { Thread t; //如果state>=0、t!=、且t没有被中断 //new Worker()时state==-1，说明不能中断 if (getState() >= 0 && (t = thread) !=  && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } }}

final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = ; // allow interrupts // new Worker()是state==-1，此处是调用Worker类的tryRelease()方法，将state置为0，而interruptIfStarted()中只有state>=0才允许调用中断 w.unlock(); 
 // 线程退出的原因，true是任务导致，false是线程正常退出 boolean completedAbruptly = true;  try { // 当前任务和从任务队列中获取的任务都为空，方停止循环 while (task !=  || (task = getTask()) != ) { //上锁可以防止在shutdown()时终止正在运行的worker，而不是应对并发 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 /** * 判断1：确保只有在线程处于stop状态且wt未中断时，wt才会被设置中断标识 * 条件1：线程池状态>=STOP,即STOP或TERMINATED * 条件2：一开始判断线程池状态<STOP，接下来检查发现Thread.interrupted()为true，即线程已经被中断，再次检查线程池状态是否>=STOP（以消除该瞬间shutdown方法生效，使线程池处于STOP或TERMINATED）， * 条件1与条件2任意满意一个，且wt不是中断状态，则中断wt，否则进入下一步 */ if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); //当前线程调用interrupt()中断
 try { //执行前（空方法，由子类重写实现） beforeExecute(wt, task);
 Throwable thrown = ; 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 = ;  w.completedTasks++; //完成任务数+1 w.unlock(); //释放锁 } } //  completedAbruptly = false; }  finally { //处理worker的退出 processWorkerExit(w, completedAbruptly); }}

private Runnable getTask() { // 最新一次poll是否超时 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. /** * 条件1：线程池状态SHUTDOWN、STOP、TERMINATED状态 * 条件2：线程池STOP、TERMINATED状态或workQueue为空 * 条件1与条件2同时为true，则workerCount-1，并且返回 * 注：条件2是考虑到SHUTDOWN状态的线程池不会接受任务，但仍会处理任务 */ if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return ; }
 int wc = workerCountOf(c);
 // Are workers subject to culling? /** * 下列两个条件满足任意一个，则给当前正在尝试获取任务的工作线程设置阻塞时间限制（超时会被销毁？不太确定这点），否则线程可以一直保持活跃状态 * 1.allowCoreThreadTimeOut：当前线程是否以keepAliveTime为超时时限等待任务 * 2.当前线程数量已经超越了核心线程数 */ boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 // 两个条件全部为true，则通过CAS使工作线程数-1，即剔除工作线程 // 条件1：工作线程数大于maximumPoolSize，或（工作线程阻塞时间受限且上次在任务队列拉取任务超时） // 条件2：wc > 1或任务队列为空 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { // 移除工作线程，成功则返回，不成功则进入下轮循环 if (compareAndDecrementWorkerCount(c)) return ; continue; }
 // 执行到这里，说明已经经过前面重重校验，开始真正获取task了 try { // 如果工作线程阻塞时间受限，则使用poll(),否则使用take() // poll()设定阻塞时间，而take()无时间限制，直到拿到结果为止 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); // r不为空，则返回该Runnable if (r != ) return r; // 没能获取到Runable，则将最近获取任务是否超时设置为true timedOut = true; } catch (InterruptedException retry) { // 响应中断，进入下一次循环前将最近获取任务超时状态置为false timedOut = false; } }}

/** * Performs cleanup and bookkeeping for a dying worker. Called * only from worker threads. Unless completedAbruptly is set, * assumes that workerCount has already been adjusted to account * for exit. This method removes thread from worker set, and * possibly terminates the pool or replaces the worker if either * it exited due to user task exception or if fewer than * corePoolSize workers are running or queue is non-empty but * there are no workers. * * @param w the worker * @param completedAbruptly if the worker died due to user exception */private void processWorkerExit(Worker w, boolean completedAbruptly) { /** * 1.工作线程-1操作 * 1）如果completedAbruptly 为true，说明工作线程发生异常，那么将正在工作的线程数量-1 * 2）如果completedAbruptly 为false，说明工作线程无任务可以执行，由getTask()执行worker-1操作 */ if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount();
 // 2.从线程set集合中移除工作线程，该过程需要加锁 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 将该worker已完成的任务数追加到线程池已完成的任务数 completedTaskCount += w.completedTasks; // HashSet<Worker>中移除该worker workers.remove(w); } finally { mainLock.unlock(); }
 // 3.根据线程池状态进行判断是否结束线程池 tryTerminate();
 /** * 4.是否需要增加工作线程 * 线程池状态是running 或 shutdown * 如果当前线程是突然终止的，addWorker() * 如果当前线程不是突然终止的，但当前线程数量 < 要维护的线程数量，addWorker() * 故如果调用线程池shutdown()，直到workQueue为空前，线程池都会维持corePoolSize个线程，然后再逐渐销毁这corePoolSize个线程 */ 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(, false); }}