ThreadPoolExecutor源码解析

悠悠 2022-03-18 10:49 211阅读 0赞

在之前章节[java并发][java]中讲过线程池，java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类，因此如果要透彻地了解Java中的线程池，必须先了解这个类。

**首先，我们带着问题去了解它**。

1. 线程池中shutdown、shutdownNow、isShutdown、isTerminated、awaitTermination的使用场景？

<table> 
 <tbody> 
  <tr> 
   <td>shutdown</td> 
   <td>在终止前允许执行以前提交的任务</td> 
  </tr> 
  <tr> 
   <td>shutdownNow</td> 
   <td>试图停止当前正执行的task，并返回尚未执行的task的list</td> 
  </tr> 
  <tr> 
   <td>isShutdown</td> 
   <td>当调用shutdown()方法后返回为true</td> 
  </tr> 
  <tr> 
   <td>isTerminated</td> 
   <td>调用shutdown()方法后，并且所有提交的任务完成后返回为true</td> 
  </tr> 
  <tr> 
   <td>awaitTermination</td> 
   <td>当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false</td> 
  </tr> 
 </tbody> 
</table>

2. ThreadPoolExecutor中参数（corePoolSize 、maximumPoolSize 、keepAliveTime、workQueue、RejectedExecutionHandler）含义？

<table> 
 <tbody> 
  <tr> 
   <td>corePoolSize</td> 
   <td> <p>核心线程会一直存活，及时没有任务需要执行；</p> <p>即使处于闲置状态也不会受<strong><code>keepAliveTime</code></strong>限制；</p> </td> 
  </tr> 
  <tr> 
   <td>maximumPoolSize</td> 
   <td> <p>最大线程数量</p> <p>注：当缓冲队列为SynchronousQueue时，它的容量为1，添加排队意义不大（缓冲队列满），而LinkedBlockingDeque时，未设置容量（<strong>默认为Integer.MAX_VALUE</strong>），maxPoolSize意义不大。</p> 
    <ul> 
     <li>当前线程数&lt;corePoolSize，创建线程，但它不会放入缓冲队列中</li> 
     <li>当前线程数=corePoolSize，且缓冲队列未满，那么任务被放入缓冲队列</li> 
     <li>当前线程数&gt;corePoolSize，且缓冲队列满，且当前线程数&lt;maxPoolSize，建新的线程来处理被添加的任务</li> 
     <li>当前线程数&gt;corePoolSize，且缓冲队列满，且当前线程数&gt;=maxPoolSize，那么通过handler所指定的策略来处理此任务。&nbsp;</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td>allowCoreThreadTimeout</td> 
   <td> <p>允许核心线程超时，默认为false</p> <p>设置为<code>true时，</code>核心线程受<strong><code>keepAliveTime</code></strong>限制的回收（则会直到线程数量=0）</p> </td> 
  </tr> 
  <tr> 
   <td>keepAliveTime</td> 
   <td> <p>线程空闲时间</p> <p>非核心线程的闲置超时时间，超过这个时间就会被回收；</p> </td> 
  </tr> 
  <tr> 
   <td>unit</td> 
   <td>指定<strong><code>keepAliveTime</code></strong>的单位，如<code>TimeUnit.SECONDS</code></td> 
  </tr> 
  <tr> 
   <td>workQueue</td> 
   <td> <p>任务队列容量（阻塞队列），常用的有三种队列：</p> <p><strong>SynchronousQueue</strong>：缓存值为1的阻塞队列</p> <p><strong>LinkedBlockingDeque</strong>：无界队列，线程池的最大线程数设置是无效的</p> </td> 
  </tr> 
  <tr> 
   <td>threadFactory</td> 
   <td>线程工厂，提供创建新线程的功能。ThreadFactory是一个接口，只有一个方法</td> 
  </tr> 
  <tr> 
   <td>rejectedExecutionHandler</td> 
   <td> <p>拒绝处理器，默认为AbortPolicy，两种情况会拒绝处理任务：</p> 
    <ul> 
     <li>当线程数=maxPoolSize，且队列已满，会拒绝新任务</li> 
     <li>当线程池被调用shutdown()后，会等待线程池里的任务执行完毕，再shutdown。如果在调用shutdown()和线程池真正shutdown之间提交任务，会拒绝新任务</li> 
    </ul><p>有几个内部实现类来处理这类情况：</p> 
    <ul> 
     <li>AbortPolicy：丢弃任务，抛出RejectedExecutionException</li> 
     <li>CallerRunsPolicy：调用者的线程会执行该任务,如果执行器已关闭,则丢弃</li> 
     <li>DiscardPolicy：忽视，什么都不会发生</li> 
     <li>DiscardOldestPolicy：从队列中踢出最先进入队列（最后一个执行）的任务</li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

**最后，读懂它的源码**

![20190220165824819.png][]

public interface Executor {
      //用来启动任务
      void execute(Runnable command);
    }
    
    public interface ExecutorService extends Executor {
        //关闭线程池，队列已经存在的任务可以继续执行
        void shutdown();
        //关闭线程池，中断未执行的任务
        List<Runnable> shutdownNow();
        //判断是否关闭
        boolean isShutdown();
        //判断是否终止
        boolean isTerminated();
        //设置超时终止
        boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException;
        //提交Callable任务
        <T> Future<T> submit(Callable<T> task);
        //提交Runable任务，带返回值
        <T> Future<T> submit(Runnable task, T result);
        //提交Runnable任务不带返回值
        Future<?> submit(Runnable task);
        //invokeAll()是同步的，其需要等待任务的完成，才能返回。submit()是异步的
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException;
    
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                      long timeout, TimeUnit unit)
            throws InterruptedException;
        //invokeAny()取第一个任务的返回值，然后调用interrupt方法中断其它任务。
        <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException;
    
        <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                        long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }
    
    public abstract class AbstractExecutorService implements ExecutorService {
    	//自定义了newTaskFor()用于构建RunnableFuture
        protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
            return new FutureTask<T>(runnable, value);
        }
    }
    
    public class ThreadPoolExecutor extends AbstractExecutorService {
        /**
         * 线程池使用一个int变量存储线程池状态和工作线程数
    	 * 为何这么玩？
         * 写c的朋友都是这么玩的，就是节省空间，int（32位，用高三位存储线程池状态，低29位存储工作线程数）
         **/
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        //COUNT_BITS=29
        private static final int COUNT_BITS = Integer.SIZE - 3;
        //理论上线程池最大线程数量CAPACITY=536870911
        private static final int CAPACITY = (1 << COUNT_BITS) - 1;
        /**
         * 线程池状态转换
         * RUNNING -> SHUTDOWN
         * RUNNING or SHUTDOWN -> STOP
         * SHUTDOWN or STOP -> TIDYING
         * TIDYING -> TERMINATED  terminated()执行完后变为该TERMINATED
         */
        //接受新任务，可以处理阻塞队列里的任务
        private static final int RUNNING = -1 << COUNT_BITS;
        //不接受新任务，可以处理阻塞队列里的任务。执行shutdown()会变为SHUTDOWN
        private static final int SHUTDOWN = 0 << COUNT_BITS;
        //不接受新的任务，不处理阻塞队列里的任务，中断正在处理的任务。执行shutdownNow()会变为STOP
        private static final int STOP = 1 << COUNT_BITS;
        //临时过渡状态，所有的任务都执行完了，当前线程池有效的线程数量为0，这个时候线程池的状态是TIDYING，执行terminated()变为TERMINATED
        private static final int TIDYING = 2 << COUNT_BITS;
        //终止状态，terminated()调用完成后的状态
        private static final int TERMINATED = 3 << COUNT_BITS;
    
        //获取线程池状态
        private static int runStateOf(int c) {
            return c & ~CAPACITY;
        }
    
        //获取工作线程数
        private static int workerCountOf(int c) {
            return c & CAPACITY;
        }
    
        //初始化ctl
        private static int ctlOf(int rs, int wc) {
            return rs | wc;
        }
    
        //用于保存等待执行的任务的阻塞队列。比如LinkedBlockQueue，SynchronousQueue等
        private final BlockingQueue<Runnable> workQueue;
        //重入锁，更新线程池核心大小、线程池最大大小等都有用到
        private final ReentrantLock mainLock = new ReentrantLock();
        //用于存储woker
        private final HashSet<Worker> workers = new HashSet<Worker>();
        //用于终止线程池
        private final Condition termination = mainLock.newCondition();
        //记录线程池中曾经出现过的最大线程数
        private int largestPoolSize;
        //完成任务数量
        private long completedTaskCount;
        //线程工厂
        private volatile ThreadFactory threadFactory;
        /**
         * rejectedExecutionHandler：任务拒绝策略
         * DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
         * AbortPolicy：抛出异常。这也是默认的策略
         * CallerRunsPolicy：用调用者所在线程来运行任务
         * DiscardPolicy：不处理，丢弃掉
         */
        private volatile RejectedExecutionHandler handler;
        /**
         * 线程空闲时间
         * 当线程空闲时间达到keepAliveTime，该线程会退出，直到线程数量等于corePoolSize。
         * 如果allowCoreThreadTimeout设置为true，则所有线程均会退出。
         */
        private volatile long keepAliveTime;
        //是否允许核心线程空闲超时退出，默认值为false。
        private volatile boolean allowCoreThreadTimeOut;
        /**
         * 核心线程数
         * 核心线程会一直存活，即使没有任务需要处理，当线程数小于核心线程数时。
         * 即使现有的线程空闲，线程池也会优先创建新线程来处理任务，而不是直接交给现有的线程处理。
         * 核心线程数在初始化时不会创建，只有提交任务的时候才会创建。核心线程在allowCoreThreadTimeout为true的时候超时会退出。
         */
        private volatile int corePoolSize;
        /**
         * 最大线程数
         * 当线程数大于或者等于核心线程，且任务队列已满时，线程池会创建新的线程，直到线程数量达到maxPoolSize。
         * 如果线程数已等于maxPoolSize，且任务队列已满，则已超出线程池的处理能力，线程池会采取拒绝操作。
         */
        private volatile int maximumPoolSize;
        //默认的拒绝策略：抛出异常
        private static final RejectedExecutionHandler defaultHandler =
                new AbortPolicy();
        private static final RuntimePermission shutdownPerm =
                new RuntimePermission("modifyThread");
    
        //构造函数
        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 == null || threadFactory == null || handler == null)
                throw new NullPointerException();
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            //转为纳秒
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }
    
        //线程池任务总数
        public long getTaskCount() {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                long n = completedTaskCount;
                for (Worker w : workers) {
                    n += w.completedTasks;
                    if (w.isLocked())
                        ++n;
                }
                return n + workQueue.size();
            } finally {
                mainLock.unlock();
            }
        }
    
        //线程池已完成的任务数量，小于等于completedTaskCount
        public long getCompletedTaskCount() {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                long n = completedTaskCount;
                for (Worker w : workers)
                    n += w.completedTasks;
                return n;
            } finally {
                mainLock.unlock();
            }
        }
    
        //线程池当前的核心线程数量
        public int getPoolSize() {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Remove rare and surprising possibility of
                // isTerminated() && getPoolSize() > 0
                return runStateAtLeast(ctl.get(), TIDYING) ? 0
                        : workers.size();
            } finally {
                mainLock.unlock();
            }
        }
    
        //线程池曾经创建过的最大线程数量
        public int getLargestPoolSize() {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                return largestPoolSize;
            } finally {
                mainLock.unlock();
            }
        }
    
        //当前线程池中正在执行任务的线程数量
        public int getActiveCount() {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                int n = 0;
                for (Worker w : workers)
                    if (w.isLocked())
                        ++n;
                return n;
            } finally {
                mainLock.unlock();
            }
        }
    
        public void execute(Runnable command) {
            //判断提交的任务是否为空
            if (command == null)
                throw new NullPointerException();
            //获取线程池状态和工作线程数量结合体(下文统称为ctl)
            int c = ctl.get();
            //判断工作线程数量是否小于核心线程数
            if (workerCountOf(c) < corePoolSize) {
                //把任务添加到worker，添加成功则返回
                if (addWorker(command, true))
                    return;
                //再次获取ctl
                c = ctl.get();
            }
            //如果线程池状态是正在运行并且添加一个任务到队列成功
            if (isRunning(c) && workQueue.offer(command)) {
                int recheck = ctl.get();
                //再次校验线程池状态，如果状态不是RUNNING则需要从队列中移除任务
                if (!isRunning(recheck) && remove(command))
                    //执行拒绝策略
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    //进入这里说明空闲核心线程数都超时退出啦
                    //因为任务已经放入队列了，所以此处不需要传入任务
                    //注意事项，网上很多说这里创建一个线程不启动这是错误的。博主亲测这里创建了一个线程并且需要启动
                    addWorker(null, false);
            }
            /*
             * 如果执行到这里有两种情况：
             * 线程池已经不是RUNNING状态
             * 线程池是RUNNING状态，workerCount>=corePoolSize并且workQueue已满
             */
            //调用addWorker(),传入false代表把线程池线程数量设置maximumPoolSize，如果添加失败则执行拒绝策略。
            else if (!addWorker(command, false))
                reject(command);
        }
    
        /**
         * @param firstTask 表示执行的任务；
         * @param core      表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断；
         *                  如果为true，根据corePoolSize来判断；如果为false，则根据maximumPoolSize来判断
         */
        private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
            for (; ; ) {
                int c = ctl.get();
                //获取线程池状态
                int rs = runStateOf(c);
                /**
                 * 只要满足下面任一条就直接返回false
                 * 线程池状态为STOP, TYDING 或 TERMINATD 状态
                 * 线程池状态为SHUTDOWN，并且firstTask != null 或者workQueue为空
                 */
                //这里为什么不直接判断线程池状态呢？是因为有可能在线程池状态为RUNNING时已经把任务放入队列中，放入完成以后状态变为SHUTDOWN
                if (rs >= SHUTDOWN && !(rs == SHUTDOWN &&
                        firstTask == null && !workQueue.isEmpty()))
                    return false;
                for (; ; ) {
                    //得到工作的线程数量
                    int wc = workerCountOf(c);
                    //如果工作线程数量大于理论上线程池容量；或者工作线程数大于(corePoolSize or maximumPoolSize) 直接返回false，添加失败
                    if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    //cas修改工作线程数，工作线程数+1。如果修改失败需要重新执行；成功退出循环
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    c = ctl.get();
                    //如果线程池状态变化则需要重新执行
                    if (runStateOf(c) != rs)
                        continue retry;
                }
            }
            //worker是否已经启动
            boolean workerStarted = false;
            //worker是否添加成功
            boolean workerAdded = false;
            Worker w = null;
            try {
                //构建worker
                w = new Worker(firstTask);
                //注意，这个thread不是firstTask，是从线程工厂造出来的
                final Thread t = w.thread;
                if (t != null) {
                    //获取锁
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
                        //获取线程池状态
                        int rs = runStateOf(ctl.get());
                        //如果线程池状态是RUNNING或者状态是SHUTDOWN但是队列里面还有任务
                        if (rs < SHUTDOWN ||
                                (rs == SHUTDOWN && firstTask == null)) {
                            //如果t.isAlive()=true，说明是有问题的，都没有启动，t怎么会是活的呢。所以抛出异常。
                            if (t.isAlive())
                                throw new IllegalThreadStateException();
                            //把worker添加到set集合
                            workers.add(w);
                            //记录线程池中出现的最大线程数
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            //添加成功标识
                            workerAdded = true;
                        }
                    } finally {
                        //释放锁
                        mainLock.unlock();
                    }
                    //添加成功启动线程，启动线程是调用了runWorker()
                    if (workerAdded) {
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
                //启动失败
                if (!workerStarted)
                    //启动失败需要从workers中移除当前构造的woker；工作线程数减1了；执行tryTerminate()判断是否终止线程池。
                    addWorkerFailed(w);
            }
            return workerStarted;
        }
    
        final void runWorker(Worker w) {
            //获取当前线程
            Thread wt = Thread.currentThread();
            //需要执行的任务
            Runnable task = w.firstTask;
            //置为null
            w.firstTask = null;
            //这里是为了把之前的state=-1设置为state=0，此时允许中断
            w.unlock();
            //是否异常退出循环
            boolean completedAbruptly = true;
            try {
                //如果任务不为null或者getTask()不为null
                while (task != null || (task = getTask()) != null) {
                    //获取锁。这里使用锁的目的在于标识正在处理任务
                    w.lock();
                    //线程池=SHUWDOWN，要保证当前线程是中断状态
                    //线程池!=SHUWDOWN，要保证当前线程不是中断状态
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                            (Thread.interrupted() &&
                                    runStateAtLeast(ctl.get(), STOP))) &&
                            !wt.isInterrupted())
                        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;
                            throw new Error(x);
                        } finally {
                            //子类执行
                            afterExecute(task, thrown);
                        }
                    } finally {
                        task = null;
                        //更新任务完成数
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                //没有出现异常
                completedAbruptly = false;
            } finally {
                //一定要注意。执行到这里说明getTask()返回null。说明当前线程池中不需要那么多线程来执行任务了，可以把多于corePoolSize数量的工作线程干掉
                processWorkerExit(w, completedAbruptly);
            }
        }
    
        //什么情况才会执行getTask呢？说明工作线程数已经大于核心线程数才会执行getTask()。一定要记住这一点
        private Runnable getTask() {
            //表示上次从阻塞队列中取任务时是否超时
            boolean timedOut = false;
            for (; ; ) {
                int c = ctl.get();
                //获取线程池状态
                int rs = runStateOf(c);
                //如果线程池状态rs >= SHUTDOWN并且(rs >= STOP，或者阻塞队列为空)。则将workerCount减1并返回null。
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                    decrementWorkerCount();
                    return null;
                }
                //获取工作线程数
                int wc = workerCountOf(c);
                //表示是否需要超时控制。allowCoreThreadTimeOut默认false；如果线程池数量超过核心线程数也是需要超时控制的
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
                //校验工作线程数量和任务队列是否为空
                if ((wc > maximumPoolSize || (timed && timedOut))
                        && (wc > 1 || workQueue.isEmpty())) {
                    //工作线程数量-1
                    if (compareAndDecrementWorkerCount(c))
                        return null;
                    continue;
                }
    
                try {
                    //根据是否超时从队列中获取任务
                    Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
                    if (r != null)
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) {
                    //说明发生了中断
                    timedOut = false;
                }
            }
        }
    
        private void processWorkerExit(Worker w, boolean completedAbruptly) {
            //是否异常，如果异常工作线程数量-1
            if (completedAbruptly)
                decrementWorkerCount();
    
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                //增加线程池完成任务数量
                completedTaskCount += w.completedTasks;
                //从worker中移除完成任务
                workers.remove(w);
            } finally {
                mainLock.unlock();
            }
            //根据线程池状态进行判断是否结束线程池
            tryTerminate();
    
            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(null, false);
            }
        }
    
        //根据线程池状态判断是否结束线程池
        final void tryTerminate() {
            for (; ; ) {
                int c = ctl.get();
                //RUNNING不能结束线程池
                //线程池状态是TIDYING或TERMINATED说明线程池已经处于正在终止的路上
                //状态为SHUTDOWN，但是任务队列不为空不能结束线程池
                if (isRunning(c) ||
                        runStateAtLeast(c, TIDYING) ||
                        (runStateOf(c) == SHUTDOWN && !workQueue.isEmpty()))
                    return;
                //工作线程数量不等于0，中断一个空闲的工作线程，并返回
                if (workerCountOf(c) != 0) {
                    interruptIdleWorkers(ONLY_ONE);
                    return;
                }
    
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // 设置线程池状态为TIDYING，如果设置成功，则调用terminated方法
                    if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                        try {
                            //子类实现
                            terminated();
                        } finally {
                            // 设置状态为TERMINATED
                            ctl.set(ctlOf(TERMINATED, 0));
                            termination.signalAll();
                        }
                        return;
                    }
                } finally {
                    mainLock.unlock();
                }
            }
        }
    
        //该方法用于给空闲工作线程一个中断标识
        private void interruptIdleWorkers(boolean onlyOne) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                //遍历worker，根据onlyOne判断，如果为ture只中断一个空闲线程
                for (Worker w : workers) {
                    Thread t = w.thread;
                    //线程没有被中断并且线程是空闲状态tryLock()判断是否空闲
                    if (!t.isInterrupted() && w.tryLock()) {
                        try {
                            t.interrupt();
                        } catch (SecurityException ignore) {
                        } finally {
                            w.unlock();
                        }
                    }
                    if (onlyOne)
                        break;
                }
            } finally {
                mainLock.unlock();
            }
        }
    }
    
    //它是ThreadPoolExecutor一个内部类，此类继AbstractQueuedSynchronizer，目的在于判断线程是否空闲以及是否可以被中断
    private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
        //工作线程空间，由线程池中所设置的线程工厂创建
        final Thread thread;
        //用来保存传入的任务
        Runnable firstTask;
        //执行完任务的总数
        volatile long completedTasks;
        Worker(Runnable firstTask) {
    		//state设置为-1是为了禁止在执行任务前对任务进行中断
            setState(-1);
    		//提交的task
            this.firstTask = firstTask;
    		//从线程工厂获取的线程，注意这个thread并不是用户线程
            this.thread = getThreadFactory().newThread(this);
        }
    
        //实现Runnable，在run()中调用了runWorker()
        public void run() {
            runWorker(this);
        }
    }

**spring的ThreadPoolTaskExecutor**

如果你对jdk的ThreadPoolExecutor感觉极度不适应的话，也可以看spring对它的封装

//org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
    ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
    //设置allowCoreThreadTimeout=true（默认false）时，核心线程会超时关闭，则会直到线程数量=0
    //默认为1
    poolTaskExecutor.setCorePoolSize(6);
    //默认为60s
    poolTaskExecutor.setKeepAliveSeconds(200);
    /**
     * 当线程数>=corePoolSize，且任务队列已满时。线程池会创建新线程来处理任务
     * 当线程数=maxPoolSize，且任务队列已满时，线程池会拒绝处理任务而抛出异常，异常见下文
     */
    //默认为Integer.MAX_VALUE
    poolTaskExecutor.setMaxPoolSize(10);
    //默认为Integer.MAX_VALUE
    poolTaskExecutor.setQueueCapacity(200);
    //默认为AbortPolicy,拒绝task的处理策略抛出RejectedExecutionException
    //poolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    poolTaskExecutor.setThreadNamePrefix("SimpleAsyncTaskExecutor-");
    //不用忘了
    poolTaskExecutor.initialize();

总结，spring使用更加简单。![java][]

[java]: /images/20220318/20f95b999654499793052b52a9f2c0f5.png
[20190220165824819.png]: /images/20220318/4aa6629a81be4bcd86b5c532184fb039.png