一篇文章带你搞懂Java线程池实现原理

/** * @author 一灯架构 * @apiNote 线程池示例 **/ public class ThreadPoolDemo { public static void main(String[] args) { // 1. 创建线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( 3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy()); // 2. 往线程池中提交3个任务 for (int i = 0; i <3; i++) { threadPoolExecutor.execute(() -> { System.out.println(Thread.currentThread().getName() + " 关注公众号：一灯架构"); }); } // 3. 关闭线程池 threadPoolExecutor.shutdown(); } } 

public class ThreadPoolExecutor extends AbstractExecutorService { // 线程池的控制状态，Integer长度是32位，前3位用来存储线程池状态，后29位用来存储线程数量 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 线程个数所占的位数 private static final int COUNT_BITS = Integer.SIZE - 3; // 线程池的最大容量，2^29-1，约5亿个线程 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // 独占锁，用来控制多线程下的并发操作 private final ReentrantLock mainLock = new ReentrantLock(); // 工作线程的集合 private final HashSet workers = new HashSet<>(); // 等待条件，用来响应中断 private final Condition termination = mainLock.newCondition(); // 是否允许回收核心线程 private volatile boolean allowCoreThreadTimeOut; // 线程数的历史峰值 private int largestPoolSize; /** * 以下是线程池的七大核心参数 */ private volatile int corePoolSize; private volatile int maximumPoolSize; private volatile long keepAliveTime; private final BlockingQueue workQueue; private volatile ThreadFactory threadFactory; private volatile RejectedExecutionHandler handler; } 

// 往线程池中提交任务 public void execute(Runnable command) { // 1. 判断提交的任务是否为null if (command == null) throw new NullPointerException(); int c = ctl.get(); // 2. 判断线程数是否小于核心线程数 if (workerCountOf(c) 

// 添加worker private boolean addWorker(Runnable firstTask, boolean core) { retry: for (; ; ) { int c = ctl.get(); int rs = runStateOf(c); // 1. 检查是否允许提交任务 if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty())) return false; // 2. 使用死循环保证添加线程成功 for (; ; ) { int wc = workerCountOf(c); // 3. 校验线程数是否超过容量限制 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 4. 使用CAS修改线程数 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // 5. 如果线程池状态变了，则从头再来 if (runStateOf(c) != rs) continue retry; } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 6. 把任务和新线程包装成一个worker w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { // 7. 加锁，控制并发 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 8. 再次校验线程池状态是否异常 int rs = runStateOf(ctl.get()); if (rs  largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 12. 启动线程 t.start(); workerStarted = true; } } } finally { if (!workerStarted) addWorkerFailed(w); } return workerStarted; } 

private final class Worker extends AbstractQueuedSynchronizer implements Runnable { // 工作线程 final Thread thread; // 任务 Runnable firstTask; // 创建worker，并创建一个新线程（用来执行任务） Worker(Runnable firstTask) { setState(-1); this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } } 

// 线程执行入口 public void run() { runWorker(this); } // 线程运行核心方法 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); boolean completedAbruptly = true; try { // 1. 如果当前worker中任务是null，就从阻塞队列中获取任务 while (task != null || (task = getTask()) != null) { // 加锁，保证thread不被其他线程中断（除非线程池被中断） w.lock(); // 2. 校验线程池状态，是否需要中断当前线程 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { // 3. 执行run方法 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 { // 4. 从worker集合删除当前worker processWorkerExit(w, completedAbruptly); } } 

// 从阻塞队列中拉取任务 private Runnable getTask() { boolean timedOut = false; for (; ; ) { int c = ctl.get(); int rs = runStateOf(c); // 1. 如果线程池已经停了，或者阻塞队列是空，就回收当前线程 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // 2. 再次判断是否需要回收线程 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { // 3. 从阻塞队列中拉取任务 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }