乐清网站制作/外链发布工具下载

        // Worker 的构造方法，主要是接受一个 task, 可以为 null, 如果非null, 将在不久的将来被执行// private final class Worker extends AbstractQueuedSynchronizer implements Runnable/*** 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 runWorkerthis.firstTask = firstTask;// 将 Worker 自身当作一个 任务，绑定到 worker.thread 中// thread 启动时，worker 就启动了this.thread = getThreadFactory().newThread(this);}// Worker 的主要工作实现，通过一个循环扫描实现        /** Delegates main run loop to outer runWorker  */public void run() {// 调用 ThreadPoolExecutor 外部实现的 runWorker 方法runWorker(this);}/*** 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*/final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 不停地从 workQueue 中获取任务，然后执行，就是这么个逻辑// getTask() 会阻塞式获取，所以 Worker 往往不会立即退出 while (task != null || (task = getTask()) != null) {// 执行过程中是不允许并发的，即同时只能一个 task 在运行，此时也不允许进行 interrupt
                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// 检测是否已被线程池是否停止 或者当前 worker 被中断// STOP = 1 << COUNT_BITS;if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())// 中断信息传递
                    wt.interrupt();try {// 任务开始前 切点，默认为空执行
                    beforeExecute(wt, task);Throwable thrown = null;try {// 直接调用任务的run方法， 具体的返回结果，会被 FutureTask 封装到 某个变量中// 可以参考以前的文章 （FutureTask是怎样获取到异步执行结果的？ https://www.cnblogs.com/yougewe/p/11666284.html）
                        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();}}// 正常退出，有必要的话，可能重新将 Worker 添加进来completedAbruptly = false;} finally {// 处理退出后下一步操作，可能重新添加 Worker
            processWorkerExit(w, completedAbruptly);}}/*** 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) {if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;workers.remove(w);} finally {mainLock.unlock();}tryTerminate();int c = ctl.get();if (runStateLessThan(c, STOP)) {// 在 Worker 正常退出的情况下，检查是否超时导致，维持最小线程数if (!completedAbruptly) {int min = allowCoreThreadTimeOut ? 0 : corePoolSize;if (min == 0 && ! workQueue.isEmpty())min = 1;// 如果满足最小线程要求，则直接返回if (workerCountOf(c) >= min)return; // replacement not needed
            }// 否则再添加一个Worker到线程池中备用// 非正常退出，会直接再添加一个WorkeraddWorker(null, false);}}/*** 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.// 如果进行了 shutdown, 且队列为空, 则需要将 worker 退出if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {// do {} while (! compareAndDecrementWorkerCount(ctl.get()));
                decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 线程数据大于最大允许线程，需要删除多余的 Workerif ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {// 如果开户了超时删除功能，则使用 poll, 否则使用 take() 进行阻塞获取Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();// 获取到任务，则可以进行执行了if (r != null)return r;// 如果有超时设置，则会在下一循环时退出timedOut = true;}// 忽略中断异常// 在这种情况下，Worker如何响应外部的中断请求呢？？？ 思考catch (InterruptedException retry) {timedOut = false;}}}

　　所以，Worker的作用就体现出来了,一个循环取任务执行任务过程：

　　　　1. 有一个主循环一直进行任务的获取;
　　　　2. 针对有超时的设置，会使用poll进行获取任务，如果超时，则 Worker 将会退出循环结束线程;
　　　　3. 无超时的设置，则会使用 take 进行阻塞式获取，直到有值;
　　　　4. 获取任务执行前置+业务+后置任务;
　　　　5. 当获取到null的任务之后，当前Worker将会结束；
　　　　6. 当前Worker结束后，将会判断是否有必要维护最低Worker数，从而决定是否再添加Worker进来。

　　还是借用一个网上同学比较通用的一个图来表述下 Worker/ThreadPoolExecutor 的工作流程吧（已经很完美，不需要再造这轮子了）

5. shutdown 操作实现

　　ThreadPoolExecutor 是通过 ctl 这个变量进行全局状态维护的，shutdown 在线程池中也是表现为一个状态，所以应该是比较简单的。

    /*** 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}*/public void shutdown() {// 为保证线程安全，使用 mainLockfinal ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// SecurityManager 检查
            checkShutdownAccess();// 设置状态为 SHUTDOWN
            advanceRunState(SHUTDOWN);// 中断空闲的 Worker, 即相当于依次关闭每个空闲线程
            interruptIdleWorkers();// 关闭钩子，默认实现为空操作，为方便子类实现自定义清理功能onShutdown(); // hook for ScheduledThreadPoolExecutor} finally {mainLock.unlock();}// 再
        tryTerminate();}/*** Transitions runState to given target, or leaves it alone if* already at least the given target.** @param targetState the desired state, either SHUTDOWN or STOP*        (but not TIDYING or TERMINATED -- use tryTerminate for that)*/private void advanceRunState(int targetState) {for (;;) {int c = ctl.get();// 自身CAS更新成功或者被其他线程更新成功if (runStateAtLeast(c, targetState) ||ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))break;}}// 关闭空闲线程（非 running 状态）/*** Common form of interruptIdleWorkers, to avoid having to* remember what the boolean argument means.*/private void interruptIdleWorkers() {// 上文已介绍， 此处 ONLY_ONE 为 false, 即是最大可能地中断所有 WorkerinterruptIdleWorkers(false);}与 shutdown 对应的，有一个 shutdownNow, 其语义是 立即停止所有任务。/*** 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();// 与 shutdown 的差别，设置的状态不一样
            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).*/private void interruptWorkers() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers)// 调用 worker 的提供的中断方法
                w.interruptIfStarted();} finally {mainLock.unlock();}}// ThreadPoolExecutor.Worker#interruptIfStartedvoid interruptIfStarted() {Thread t;if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {try {// 直接调用任务的 interrupt
                    t.interrupt();} catch (SecurityException ignore) {}}}

    // invokeAll 的方法直接在抽象方便中就实现了，它的语义是同时执行n个任务，并同步等待结果返回// java.util.concurrent.AbstractExecutorService#invokeAll(java.util.Collection<? extends java.util.concurrent.Callable<T>>)public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)throws InterruptedException {if (tasks == null)throw new NullPointerException();ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());boolean done = false;try {for (Callable<T> t : tasks) {RunnableFuture<T> f = newTaskFor(t);futures.add(f);// 依次调用各子类的实现，添加任务
                execute(f);}for (int i = 0, size = futures.size(); i < size; i++) {Future<T> f = futures.get(i);if (!f.isDone()) {try {// 依次等待执行结果
                        f.get();} catch (CancellationException ignore) {} catch (ExecutionException ignore) {}}}done = true;return futures;} finally {if (!done)for (int i = 0, size = futures.size(); i < size; i++)futures.get(i).cancel(true);}}

    // runState is stored in the high-order bits// 整个状态使值使用 ctl 的高三位值进行控制， COUNT_BITS=29// 1110 0000 0000 0000private static final int RUNNING    = -1 << COUNT_BITS;// 0000 0000 0000 0000private static final int SHUTDOWN   =  0 << COUNT_BITS;// 0010 0000 0000 0000private static final int STOP       =  1 << COUNT_BITS;// 0100 0000 0000 0000private static final int TIDYING    =  2 << COUNT_BITS;// 0110 0000 0000 0000private static final int TERMINATED =  3 << COUNT_BITS;// 整个状态值的大小顺序主: RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED// 而低 29位，则用来保存 worker 的数量，当worker增加时，只要将整个 ctl 增加即可。// 0001 1111 1111 1111, 即是最大的 worker 数量private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// 整个 ctl 描述为一个 AtomicInteger, 功能如下:/*** 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).*/private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    // ThreadPoolExecutor.awaitTermination()public boolean awaitTermination(long timeout, TimeUnit unit)throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (;;) {// 只是循环 ctl 状态, 只要 状态为 TERMINATED 状态，则说明已经关闭成功// 此处 termination 的状态触发是在 tryTerminate 中触发的if (runStateAtLeast(ctl.get(), TERMINATED))return true;if (nanos <= 0)return false;nanos = termination.awaitNanos(nanos);}} finally {mainLock.unlock();}}