// 1. `ctl`，可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 2. `COUNT_BITS`，`Integer.SIZE`为32，所以`COUNT_BITS`为29 private static final int COUNT_BITS = Integer.SIZE - 3; // 3. `CAPACITY`，线程池允许的最大线程数。1左移29位，然后减1，即为 2^29 - 1，二进制表示为连续的29个1 private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits // 4. 线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED . 此处RUNNING实际上是-(1<<COUNT_BITS)，是负数。 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;
// Packing and unpacking ctl // 5. `runStateOf()`，获取线程池状态，通过按位与操作，低29位将全部变成0 private static int runStateOf(int c) { return c & ~CAPACITY; } // 6. `workerCountOf()`，获取线程池worker数量，通过按位与操作，高3位将全部变成0 private static int workerCountOf(int c) { return c & CAPACITY; } // 7. `ctlOf()`，根据线程池状态和线程池worker数量，生成ctl值 private static int ctlOf(int rs, int wc) { return rs | wc; }
/* * Bit field accessors that don't require unpacking ctl. * These depend on the bit layout and on workerCount being never negative. */ // 8. `runStateLessThan()`，线程池状态小于xx private static boolean runStateLessThan(int c, int s) { return c < s; } // 9. `runStateAtLeast()`，线程池状态大于等于xx private static boolean runStateAtLeast(int c, int s) { return c >= s; }
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); // worker数量比核心线程数小，直接创建worker执行任务 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // worker数量超过核心线程数，任务直接进入队列 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); // 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。 // 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。 if (! isRunning(recheck) && remove(command)) reject(command); // 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0 else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。 // 这儿有3点需要注意： // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态 // 2. addWorker第2个参数表示是否创建核心线程 // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作 else if (!addWorker(command, false)) reject(command); }
// 内层自旋 for (;;) { int wc = workerCountOf(c); // worker数量超过容量，直接返回false if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 使用CAS的方式增加worker数量。 // 若增加成功，则直接跳出外层循环进入到第二部分 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl // 线程池状态发生变化，对外层循环进行自旋 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); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; // worker的添加必须是串行的，因此需要加锁 mainLock.lock(); try { // 这儿需要重新检查线程池状态 // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { // worker已经调用过了start()方法，则不再创建worker if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); // worker创建并添加到workers成功 workers.add(w); // 更新`largestPoolSize`变量 int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } // 启动worker线程 if (workerAdded) { t.start(); workerStarted = true; } } } finally { // worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作 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. 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 runWorker this.firstTask = firstTask; // 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker this.thread = getThreadFactory().newThread(this); }
/** Delegates main run loop to outer runWorker */ public void run() { //此处分析参见 `核心线程执行逻辑-runworker` runWorker(this); }