/** * The main pool control state, ctl, is an atomic integer packing * two conceptual fields * workerCount, indicating the effective number of threads * runState, indicating whether running, shutting down etc * * In order to pack them into one int, we limit workerCount to * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2 * billion) otherwise representable. If this is ever an issue in * the future, the variable can be changed to be an AtomicLong, * and the shift/mask constants below adjusted. But until the need * arises, this code is a bit faster and simpler using an int. * * The workerCount is the number of workers that have been * permitted to start and not permitted to stop. The value may be * transiently different from the actual number of live threads, * for example when a ThreadFactory fails to create a thread when * asked, and when exiting threads are still performing * bookkeeping before terminating. The user-visible pool size is * reported as the current size of the workers set. * * The runState provides the main lifecycle control, taking on values: * * RUNNING: Accept new tasks and process queued tasks * SHUTDOWN: Don't accept new tasks, but process queued tasks * STOP: Don't accept new tasks, don't process queued tasks, * and interrupt in-progress tasks * TIDYING: All tasks have terminated, workerCount is zero, * the thread transitioning to state TIDYING * will run the terminated() hook method * TERMINATED: terminated() has completed * * The numerical order among these values matters, to allow * ordered comparisons. The runState monotonically increases over * time, but need not hit each state. The transitions are: * * RUNNING -> SHUTDOWN * On invocation of shutdown(), perhaps implicitly in finalize() * (RUNNING or SHUTDOWN) -> STOP * On invocation of shutdownNow() * SHUTDOWN -> TIDYING * When both queue and pool are empty * STOP -> TIDYING * When pool is empty * TIDYING -> TERMINATED * When the terminated() hook method has completed * * Threads waiting in awaitTermination() will return when the * state reaches TERMINATED. * * Detecting the transition from SHUTDOWN to TIDYING is less * straightforward than you'd like because the queue may become * empty after non-empty and vice versa during SHUTDOWN state, but * we can only terminate if, after seeing that it is empty, we see * that workerCount is 0 (which sometimes entails a recheck -- see * below). */ privatefinal AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); privatestaticfinalint COUNT_BITS = Integer.SIZE - 3; privatestaticfinalint CAPACITY = (1 << COUNT_BITS) - 1;
/** * Executes the given task sometime in the future. The task * may execute in a new thread or in an existing pooled thread. * * If the task cannot be submitted for execution, either because this * executor has been shutdown or because its capacity has been reached, * the task is handled by the current {@code RejectedExecutionHandler}. * * @param command the task to execute * @throws RejectedExecutionException at discretion of * {@code RejectedExecutionHandler}, if the task * cannot be accepted for execution * @throws NullPointerException if {@code command} is null */ publicvoidexecute(Runnable command){ if (command == null) thrownew 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(); // 首先它会获取当前内部数据 ctl，然后从中提取工作线程数（后29位），如果小于 corePoolSize，则创建新的线程 if (workerCountOf(c) < corePoolSize) { // 线程数小于核心线程数时任务不入队，直接通过参数传递到 worker 线程中 if (addWorker(command, true)) // 创建成功的话，直接返回 return; // 拉取最新内部数据 ctl c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { // 如果当前还处于运行状态，并且任务队列没满 int recheck = ctl.get(); // 因为查看状态时并没有用到锁，所以这里在检查一次如果当前状态不是运行中，就把任务从队列中删除，然后拒绝该任务 if (! isRunning(recheck) && remove(command)) reject(command); // 否则，检查工作线程数是否为 0，是的话就添加一个工作线程，之所以会出现这种情况是因为在 workQueue.offer 执行之前可能最后一个线程被销毁了（考虑 keepAliveTime） elseif (workerCountOf(recheck) == 0) addWorker(null, false); } // 如果当前状态不是运行中，或者工作线程满了，入队失败时，当前任务也是参数传递到 worker 线程中 elseif (!addWorker(command, false)) // 如果 addWorker 失败，既有可能是线程池已经停止，也有可能是线程数达到 maxThreadSize，无论是哪一种，都需要拒绝该任务 reject(command); // 如果 addWorker 成功了，说明当前状态还是运行中，是工作队列满了，但是线程数没有达到 maxThreadSize }
/** * Checks if a new worker can be added with respect to current * pool state and the given bound (either core or maximum). If so, * the worker count is adjusted accordingly, and, if possible, a * new worker is created and started, running firstTask as its * first task. This method returns false if the pool is stopped or * eligible to shut down. It also returns false if the thread * factory fails to create a thread when asked. If the thread * creation fails, either due to the thread factory returning * null, or due to an exception (typically OutOfMemoryError in * Thread.start()), we roll back cleanly. * * @param firstTask the task the new thread should run first (or * null if none). Workers are created with an initial first task * (in method execute()) to bypass queuing when there are fewer * than corePoolSize threads (in which case we always start one), * or when the queue is full (in which case we must bypass queue). * Initially idle threads are usually created via * prestartCoreThread or to replace other dying workers. * * @param core if true use corePoolSize as bound, else * maximumPoolSize. (A boolean indicator is used here rather than a * value to ensure reads of fresh values after checking other pool * state). * @return true if successful */ privatebooleanaddWorker(Runnable firstTask, boolean core){ retry: for (;;) { // 循环 CAS int c = ctl.get(); int rs = runStateOf(c);
// Check if queue empty only if necessary. // 如果调用过 shutdown 或者 shutdownNow 一般就不用创建工作线程了，但是这里要排除一种情况： // 考虑到当前状态是 SHUTDOWN，并且 firstTask == null，就是execute中addWorker(null, false)的情况，这说明刚才有一个任务已经入队了，但是最后一个工作线程可能在 workQueue.offer 执行之前被销毁了 // 而且 !workQueue.isEmpty() 工作队列不等于空，这时候就需要创建一个线程来吧 workQueue 中剩下的任务处理完 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) returnfalse; // 通过了状态检查，这里我们通过 CAS 修改工作线程数 for (;;) { int wc = workerCountOf(c); // 检查线程数是否过多 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) returnfalse; // cas 操作成功则跳到下一步 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 } } // 走到这一步说明工作线程数已经成功+1，状态目前来看没问题 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 创建工作线程 w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { // 这里使用到了一个锁 mainLock，它主要是用来保护所有工作线程的集合 workers，而且在执行shutdown 时也会持有该锁，所以这里在锁的保护下进行最终的状态确认 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); // 最后的状态检查，并加入到 workers 集合中，小于 SHUTDOWN 说明当前状态是 RUNING，或者 rs == SHUTDOWN && firstTask == null，这就是execute中addWorker(null, false)的情况 if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable thrownew 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 { // 否则，通过 CAS 进行工作线程数 -1，检查终止状态，并协助最终的状态转换 // 前两个操作比较好理解，这里所谓的状态转换是指从 SHUTDOWN -> TIDYING->TERMINATED的转换 // 在 SHUTDOWN 时，如果发现线程数为 0了，就开始状态转换 if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
/** * Rolls back the worker thread creation. * - removes worker from workers, if present * - decrements worker count * - rechecks for termination, in case the existence of this * worker was holding up termination */ privatevoidaddWorkerFailed(Worker w){ final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) workers.remove(w); decrementWorkerCount(); // 检查当前的线程池状态和工作线程数，如果任务队列空了，并且线程数为0，就开始执行 terminated 回调 tryTerminate(); } finally { mainLock.unlock(); } }
/** * Class Worker mainly maintains interrupt control state for * threads running tasks, along with other minor bookkeeping. * This class opportunistically extends AbstractQueuedSynchronizer * to simplify acquiring and releasing a lock surrounding each * task execution. This protects against interrupts that are * intended to wake up a worker thread waiting for a task from * instead interrupting a task being run. We implement a simple * non-reentrant mutual exclusion lock rather than use * ReentrantLock because we do not want worker tasks to be able to * reacquire the lock when they invoke pool control methods like * setCorePoolSize. Additionally, to suppress interrupts until * the thread actually starts running tasks, we initialize lock * state to a negative value, and clear it upon start (in * runWorker). */ privatefinalclassWorker extendsAbstractQueuedSynchronizer implementsRunnable { /** 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 */ volatilelong 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; this.thread = getThreadFactory().newThread(this); }
/** Delegates main run loop to outer runWorker */ publicvoidrun(){ runWorker(this); }
// Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state.
/** * Main worker run loop. Repeatedly gets tasks from queue and * executes them, while coping with a number of issues: * * 1. We may start out with an initial task, in which case we * don't need to get the first one. Otherwise, as long as pool is * running, we get tasks from getTask. If it returns null then the * worker exits due to changed pool state or configuration * parameters. Other exits result from exception throws in * external code, in which case completedAbruptly holds, which * usually leads processWorkerExit to replace this thread. * * 2. Before running any task, the lock is acquired to prevent * other pool interrupts while the task is executing, and then we * ensure that unless pool is stopping, this thread does not have * its interrupt set. * * 3. Each task run is preceded by a call to beforeExecute, which * might throw an exception, in which case we cause thread to die * (breaking loop with completedAbruptly true) without processing * the task. * * 4. Assuming beforeExecute completes normally, we run the task, * gathering any of its thrown exceptions to send to afterExecute. * We separately handle RuntimeException, Error (both of which the * specs guarantee that we trap) and arbitrary Throwables. * Because we cannot rethrow Throwables within Runnable.run, we * wrap them within Errors on the way out (to the thread's * UncaughtExceptionHandler). Any thrown exception also * conservatively causes thread to die. * * 5. After task.run completes, we call afterExecute, which may * also throw an exception, which will also cause thread to * die. According to JLS Sec 14.20, this exception is the one that * will be in effect even if task.run throws. * * The net effect of the exception mechanics is that afterExecute * and the thread's UncaughtExceptionHandler have as accurate * information as we can provide about any problems encountered by * user code. * * @param w the worker */ finalvoidrunWorker(Worker w){ Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; // 解锁，此后 shutdown 函数才能中断该线程 w.unlock(); // allow interrupts boolean completedAbruptly = true; try { // 如果 firstTask 不等于 null，则不检查状态就开始试图执行 while (task != null || (task = getTask()) != null) { // getTask 中会检查状态，并从任务队列中拉取任务 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 // 这个地方设计的比较复杂，就像前面说的如果执行 firstTask 时被SHUTDOWN，那么这个没有进入任务队列的 firstTask 是需要正常执行完的， // 但是 firstTask 执行前可能线程被中断了（调用了shutdown函数），这时候我们需要清除中断标志位才行，这样才能算正常执行，也就是下面的第一次 Thread.interrupted() 调用 // 而当我们执行 shutdownNow 时，线程池的状态是 STOP，又或者在我们刚才进行清除中断标志位之后线程池的状态变成了 STOP，并且当前中断标志位没有被有效的设置的话 !wt.isInterrupted()， // 我们就要补上刚才误清的中断标志位，注意这里我们并没有直接结束线程，而是设置标志位并执行目标任务，让目标任务去决定遇到中断标志位时需要作出什么处理，而不是线程池直接掌管生杀大权 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) // 如果状态已经是 STOP，就打断自己, 只会设置中断标志位，任务还是会继续执行的 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; thrownew Error(x); } finally { // 执行回调函数 afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
/** * Performs blocking or timed wait for a task, depending on * current configuration settings, or returns null if this worker * must exit because of any of: * 1. There are more than maximumPoolSize workers (due to * a call to setMaximumPoolSize). * 2. The pool is stopped. * 3. The pool is shutdown and the queue is empty. * 4. This worker timed out waiting for a task, and timed-out * workers are subject to termination (that is, * {@code allowCoreThreadTimeOut || workerCount > corePoolSize}) * both before and after the timed wait, and if the queue is * non-empty, this worker is not the last thread in the pool. * * @return task, or null if the worker must exit, in which case * workerCount is decremented */ private Runnable getTask(){ 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. // 如果线程池已经STOP或者 SHUTDOWN状态时任务队列为空，就销毁线程 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); returnnull; }
int wc = workerCountOf(c);
// Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 如果线程数过多或者超时，并且当前线程数>1或者任务队列为空，就销毁线程 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) returnnull; continue; }
/** * 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 */ privatevoidprocessWorkerExit(Worker w, boolean completedAbruptly){ if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount();
/** * Initiates an orderly shutdown in which previously submitted * tasks are executed, but no new tasks will be accepted. * Invocation has no additional effect if already shut down. * * <p>This method does not wait for previously submitted tasks to * complete execution. Use {@link #awaitTermination awaitTermination} * to do that. * * @throws SecurityException {@inheritDoc} */ publicvoidshutdown(){ final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(SHUTDOWN); interruptIdleWorkers(); onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } tryTerminate(); }
/** * Interrupts threads that might be waiting for tasks (as * indicated by not being locked) so they can check for * termination or configuration changes. Ignores * SecurityExceptions (in which case some threads may remain * uninterrupted). * * @param onlyOne If true, interrupt at most one worker. This is * called only from tryTerminate when termination is otherwise * enabled but there are still other workers. In this case, at * most one waiting worker is interrupted to propagate shutdown * signals in case all threads are currently waiting. * Interrupting any arbitrary thread ensures that newly arriving * workers since shutdown began will also eventually exit. * To guarantee eventual termination, it suffices to always * interrupt only one idle worker, but shutdown() interrupts all * idle workers so that redundant workers exit promptly, not * waiting for a straggler task to finish. */ privatevoidinterruptIdleWorkers(boolean onlyOne){ // mainLock 保护 workers 集合 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(); } }
/** * Attempts to stop all actively executing tasks, halts the * processing of waiting tasks, and returns a list of the tasks * that were awaiting execution. These tasks are drained (removed) * from the task queue upon return from this method. * * <p>This method does not wait for actively executing tasks to * terminate. Use {@link #awaitTermination awaitTermination} to * do that. * * <p>There are no guarantees beyond best-effort attempts to stop * processing actively executing tasks. This implementation * cancels tasks via {@link Thread#interrupt}, so any task that * fails to respond to interrupts may never terminate. * * @throws SecurityException {@inheritDoc} */ public List<Runnable> shutdownNow(){ List<Runnable> tasks; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { checkShutdownAccess(); advanceRunState(STOP); interruptWorkers(); tasks = drainQueue(); } finally { mainLock.unlock(); } tryTerminate(); return tasks; }
/** * Interrupts all threads, even if active. Ignores SecurityExceptions * (in which case some threads may remain uninterrupted). */ privatevoidinterruptWorkers(){ final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) w.interruptIfStarted(); } finally { mainLock.unlock(); } }
最后我们简单说一说 ScheduledThreadPoolExecutor，它是在 ThreadPoolExecutor 的基础上额外实现了定时任务的功能，你可以简单地认为它的核心内容就是实现了一个 ThreadPoolExecutor 中使用到的阻塞队列，ScheduledThreadPoolExecutor 的内部实现了一个延时工作队列, 队列中的任务按照执行时间点排序(使用到二分查找)，当第一个工作线程发现当前没有可执行的任务（下一个任务可能要在 N 秒之后执行）时，它会成为领头人线程，而其他线程都会在一个 Condition 上永久等待，而领头人线程会在该 Condition 上等待 N 秒，当等待超时或者有需要立刻执行的任务被添加时，领头人线程会苏醒过来并唤醒该 Condition 上的下一个线程。通过这个延时工作队列，ScheduledThreadPoolExecutor 达到了没有任务到期时，所有工作线程都在等待。有任务到期时，线程会立马唤醒并开始工作的效果。
