ThreadPoolExecutor源码解析

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

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

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

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























shutdown 在终止前允许执行以前提交的任务
shutdownNow 试图停止当前正执行的task，并返回尚未执行的task的list
isShutdown 当调用shutdown()方法后返回为true
isTerminated 调用shutdown()方法后，并且所有提交的任务完成后返回为true
awaitTermination 当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false
  1. ThreadPoolExecutor中参数（corePoolSize 、maximumPoolSize 、keepAliveTime、workQueue、RejectedExecutionHandler）含义？



































corePoolSize

核心线程会一直存活，及时没有任务需要执行；

即使处于闲置状态也不会受keepAliveTime限制；
maximumPoolSize

最大线程数量

注：当缓冲队列为SynchronousQueue时，它的容量为1，添加排队意义不大（缓冲队列满），而LinkedBlockingDeque时，未设置容量（默认为Integer.MAX_VALUE），maxPoolSize意义不大。



  • 当前线程数<corePoolSize，创建线程，但它不会放入缓冲队列中

  • 当前线程数=corePoolSize，且缓冲队列未满，那么任务被放入缓冲队列

  • 当前线程数>corePoolSize，且缓冲队列满，且当前线程数<maxPoolSize，建新的线程来处理被添加的任务

  • 当前线程数>corePoolSize，且缓冲队列满，且当前线程数>=maxPoolSize，那么通过handler所指定的策略来处理此任务。 
allowCoreThreadTimeout

允许核心线程超时，默认为false

设置为true时，核心线程受keepAliveTime限制的回收（则会直到线程数量=0）
keepAliveTime

线程空闲时间

非核心线程的闲置超时时间，超过这个时间就会被回收；
unit 指定keepAliveTime的单位，如TimeUnit.SECONDS
workQueue

任务队列容量（阻塞队列），常用的有三种队列：

SynchronousQueue：缓存值为1的阻塞队列

LinkedBlockingDeque：无界队列，线程池的最大线程数设置是无效的
threadFactory 线程工厂，提供创建新线程的功能。ThreadFactory是一个接口，只有一个方法
rejectedExecutionHandler

拒绝处理器，默认为AbortPolicy，两种情况会拒绝处理任务：



  • 当线程数=maxPoolSize，且队列已满，会拒绝新任务

  • 当线程池被调用shutdown()后，会等待线程池里的任务执行完毕，再shutdown。如果在调用shutdown()和线程池真正shutdown之间提交任务，会拒绝新任务

有几个内部实现类来处理这类情况：



  • AbortPolicy：丢弃任务，抛出RejectedExecutionException

  • CallerRunsPolicy：调用者的线程会执行该任务,如果执行器已关闭,则丢弃

  • DiscardPolicy：忽视，什么都不会发生

  • DiscardOldestPolicy：从队列中踢出最先进入队列（最后一个执行）的任务

最后，读懂它的源码

20190220165824819.png

  1. public interface Executor {
  2.   //用来启动任务
  3.   void execute(Runnable command);
  4. }
  6. public interface ExecutorService extends Executor {
  7.     //关闭线程池，队列已经存在的任务可以继续执行
  8.     void shutdown();
  9.     //关闭线程池，中断未执行的任务
  10.     List<Runnable> shutdownNow();
  11.     //判断是否关闭
  12.     boolean isShutdown();
  13.     //判断是否终止
  14.     boolean isTerminated();
  15.     //设置超时终止
  16.     boolean awaitTermination(long timeout, TimeUnit unit)
  17.         throws InterruptedException;
  18.     //提交Callable任务
  19.     <T> Future<T> submit(Callable<T> task);
  20.     //提交Runable任务，带返回值
  21.     <T> Future<T> submit(Runnable task, T result);
  22.     //提交Runnable任务不带返回值
  23.     Future<?> submit(Runnable task);
  24.     //invokeAll()是同步的，其需要等待任务的完成，才能返回。submit()是异步的
  25.     <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
  26.         throws InterruptedException;
  28.     <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
  29.                                   long timeout, TimeUnit unit)
  30.         throws InterruptedException;
  31.     //invokeAny()取第一个任务的返回值，然后调用interrupt方法中断其它任务。
  32.     <T> T invokeAny(Collection<? extends Callable<T>> tasks)
  33.         throws InterruptedException, ExecutionException;
  35.     <T> T invokeAny(Collection<? extends Callable<T>> tasks,
  36.                     long timeout, TimeUnit unit)
  37.         throws InterruptedException, ExecutionException, TimeoutException;
  38. }
  40. public abstract class AbstractExecutorService implements ExecutorService {
  41.     //自定义了newTaskFor()用于构建RunnableFuture
  42.     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
  43.         return new FutureTask<T>(runnable, value);
  44.     }
  45. }
  47. public class ThreadPoolExecutor extends AbstractExecutorService {
  48.     /**
  49.      * 线程池使用一个int变量存储线程池状态和工作线程数
  50.      * 为何这么玩？
  51.      * 写c的朋友都是这么玩的，就是节省空间，int（32位，用高三位存储线程池状态，低29位存储工作线程数）
  52.      **/
  53.     private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  54.     //COUNT_BITS=29
  55.     private static final int COUNT_BITS = Integer.SIZE - 3;
  56.     //理论上线程池最大线程数量CAPACITY=536870911
  57.     private static final int CAPACITY = (1 << COUNT_BITS) - 1;
  58.     /**
  59.      * 线程池状态转换
  60.      * RUNNING -> SHUTDOWN
  61.      * RUNNING or SHUTDOWN -> STOP
  62.      * SHUTDOWN or STOP -> TIDYING
  63.      * TIDYING -> TERMINATED  terminated()执行完后变为该TERMINATED
  64.      */
  65.     //接受新任务，可以处理阻塞队列里的任务
  66.     private static final int RUNNING = -1 << COUNT_BITS;
  67.     //不接受新任务，可以处理阻塞队列里的任务。执行shutdown()会变为SHUTDOWN
  68.     private static final int SHUTDOWN = 0 << COUNT_BITS;
  69.     //不接受新的任务，不处理阻塞队列里的任务，中断正在处理的任务。执行shutdownNow()会变为STOP
  70.     private static final int STOP = 1 << COUNT_BITS;
  71.     //临时过渡状态，所有的任务都执行完了，当前线程池有效的线程数量为0，这个时候线程池的状态是TIDYING，执行terminated()变为TERMINATED
  72.     private static final int TIDYING = 2 << COUNT_BITS;
  73.     //终止状态，terminated()调用完成后的状态
  74.     private static final int TERMINATED = 3 << COUNT_BITS;
  76.     //获取线程池状态
  77.     private static int runStateOf(int c) {
  78.         return c & ~CAPACITY;
  79.     }
  81.     //获取工作线程数
  82.     private static int workerCountOf(int c) {
  83.         return c & CAPACITY;
  84.     }
  86.     //初始化ctl
  87.     private static int ctlOf(int rs, int wc) {
  88.         return rs | wc;
  89.     }
  91.     //用于保存等待执行的任务的阻塞队列。比如LinkedBlockQueue，SynchronousQueue等
  92.     private final BlockingQueue<Runnable> workQueue;
  93.     //重入锁，更新线程池核心大小、线程池最大大小等都有用到
  94.     private final ReentrantLock mainLock = new ReentrantLock();
  95.     //用于存储woker
  96.     private final HashSet<Worker> workers = new HashSet<Worker>();
  97.     //用于终止线程池
  98.     private final Condition termination = mainLock.newCondition();
  99.     //记录线程池中曾经出现过的最大线程数
  100.     private int largestPoolSize;
  101.     //完成任务数量
  102.     private long completedTaskCount;
  103.     //线程工厂
  104.     private volatile ThreadFactory threadFactory;
  105.     /**
  106.      * rejectedExecutionHandler：任务拒绝策略
  107.      * DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
  108.      * AbortPolicy：抛出异常。这也是默认的策略
  109.      * CallerRunsPolicy：用调用者所在线程来运行任务
  110.      * DiscardPolicy：不处理，丢弃掉
  111.      */
  112.     private volatile RejectedExecutionHandler handler;
  113.     /**
  114.      * 线程空闲时间
  115.      * 当线程空闲时间达到keepAliveTime，该线程会退出，直到线程数量等于corePoolSize。
  116.      * 如果allowCoreThreadTimeout设置为true，则所有线程均会退出。
  117.      */
  118.     private volatile long keepAliveTime;
  119.     //是否允许核心线程空闲超时退出，默认值为false。
  120.     private volatile boolean allowCoreThreadTimeOut;
  121.     /**
  122.      * 核心线程数
  123.      * 核心线程会一直存活，即使没有任务需要处理，当线程数小于核心线程数时。
  124.      * 即使现有的线程空闲，线程池也会优先创建新线程来处理任务，而不是直接交给现有的线程处理。
  125.      * 核心线程数在初始化时不会创建，只有提交任务的时候才会创建。核心线程在allowCoreThreadTimeout为true的时候超时会退出。
  126.      */
  127.     private volatile int corePoolSize;
  128.     /**
  129.      * 最大线程数
  130.      * 当线程数大于或者等于核心线程，且任务队列已满时，线程池会创建新的线程，直到线程数量达到maxPoolSize。
  131.      * 如果线程数已等于maxPoolSize，且任务队列已满，则已超出线程池的处理能力，线程池会采取拒绝操作。
  132.      */
  133.     private volatile int maximumPoolSize;
  134.     //默认的拒绝策略：抛出异常
  135.     private static final RejectedExecutionHandler defaultHandler =
  136.             new AbortPolicy();
  137.     private static final RuntimePermission shutdownPerm =
  138.             new RuntimePermission("modifyThread");
  140.     //构造函数
  141.     public ThreadPoolExecutor(int corePoolSize,
  142.                               int maximumPoolSize,
  143.                               long keepAliveTime,
  144.                               TimeUnit unit,
  145.                               BlockingQueue<Runnable> workQueue,
  146.                               ThreadFactory threadFactory,
  147.                               RejectedExecutionHandler handler) {
  148.         if (corePoolSize < 0 ||
  149.                 maximumPoolSize <= 0 ||
  150.                 maximumPoolSize < corePoolSize ||
  151.                 keepAliveTime < 0)
  152.             throw new IllegalArgumentException();
  153.         if (workQueue == null || threadFactory == null || handler == null)
  154.             throw new NullPointerException();
  155.         this.corePoolSize = corePoolSize;
  156.         this.maximumPoolSize = maximumPoolSize;
  157.         this.workQueue = workQueue;
  158.         //转为纳秒
  159.         this.keepAliveTime = unit.toNanos(keepAliveTime);
  160.         this.threadFactory = threadFactory;
  161.         this.handler = handler;
  162.     }
  164.     //线程池任务总数
  165.     public long getTaskCount() {
  166.         final ReentrantLock mainLock = this.mainLock;
  167.         mainLock.lock();
  168.         try {
  169.             long n = completedTaskCount;
  170.             for (Worker w : workers) {
  171.                 n += w.completedTasks;
  172.                 if (w.isLocked())
  173.                     ++n;
  174.             }
  175.             return n + workQueue.size();
  176.         } finally {
  177.             mainLock.unlock();
  178.         }
  179.     }
  181.     //线程池已完成的任务数量，小于等于completedTaskCount
  182.     public long getCompletedTaskCount() {
  183.         final ReentrantLock mainLock = this.mainLock;
  184.         mainLock.lock();
  185.         try {
  186.             long n = completedTaskCount;
  187.             for (Worker w : workers)
  188.                 n += w.completedTasks;
  189.             return n;
  190.         } finally {
  191.             mainLock.unlock();
  192.         }
  193.     }
  195.     //线程池当前的核心线程数量
  196.     public int getPoolSize() {
  197.         final ReentrantLock mainLock = this.mainLock;
  198.         mainLock.lock();
  199.         try {
  200.             // Remove rare and surprising possibility of
  201.             // isTerminated() && getPoolSize() > 0
  202.             return runStateAtLeast(ctl.get(), TIDYING) ? 0
  203.                     : workers.size();
  204.         } finally {
  205.             mainLock.unlock();
  206.         }
  207.     }
  209.     //线程池曾经创建过的最大线程数量
  210.     public int getLargestPoolSize() {
  211.         final ReentrantLock mainLock = this.mainLock;
  212.         mainLock.lock();
  213.         try {
  214.             return largestPoolSize;
  215.         } finally {
  216.             mainLock.unlock();
  217.         }
  218.     }
  220.     //当前线程池中正在执行任务的线程数量
  221.     public int getActiveCount() {
  222.         final ReentrantLock mainLock = this.mainLock;
  223.         mainLock.lock();
  224.         try {
  225.             int n = 0;
  226.             for (Worker w : workers)
  227.                 if (w.isLocked())
  228.                     ++n;
  229.             return n;
  230.         } finally {
  231.             mainLock.unlock();
  232.         }
  233.     }
  235.     public void execute(Runnable command) {
  236.         //判断提交的任务是否为空
  237.         if (command == null)
  238.             throw new NullPointerException();
  239.         //获取线程池状态和工作线程数量结合体(下文统称为ctl)
  240.         int c = ctl.get();
  241.         //判断工作线程数量是否小于核心线程数
  242.         if (workerCountOf(c) < corePoolSize) {
  243.             //把任务添加到worker，添加成功则返回
  244.             if (addWorker(command, true))
  245.                 return;
  246.             //再次获取ctl
  247.             c = ctl.get();
  248.         }
  249.         //如果线程池状态是正在运行并且添加一个任务到队列成功
  250.         if (isRunning(c) && workQueue.offer(command)) {
  251.             int recheck = ctl.get();
  252.             //再次校验线程池状态，如果状态不是RUNNING则需要从队列中移除任务
  253.             if (!isRunning(recheck) && remove(command))
  254.                 //执行拒绝策略
  255.                 reject(command);
  256.             else if (workerCountOf(recheck) == 0)
  257.                 //进入这里说明空闲核心线程数都超时退出啦
  258.                 //因为任务已经放入队列了，所以此处不需要传入任务
  259.                 //注意事项，网上很多说这里创建一个线程不启动这是错误的。博主亲测这里创建了一个线程并且需要启动
  260.                 addWorker(null, false);
  261.         }
  262.         /*
  263.          * 如果执行到这里有两种情况：
  264.          * 线程池已经不是RUNNING状态
  265.          * 线程池是RUNNING状态，workerCount>=corePoolSize并且workQueue已满
  266.          */
  267.         //调用addWorker(),传入false代表把线程池线程数量设置maximumPoolSize，如果添加失败则执行拒绝策略。
  268.         else if (!addWorker(command, false))
  269.             reject(command);
  270.     }
  272.     /**
  273.      * @param firstTask 表示执行的任务；
  274.      * @param core      表示限制添加线程的数量是根据corePoolSize来判断还是maximumPoolSize来判断；
  275.      *                  如果为true，根据corePoolSize来判断；如果为false，则根据maximumPoolSize来判断
  276.      */
  277.     private boolean addWorker(Runnable firstTask, boolean core) {
  278.         retry:
  279.         for (; ; ) {
  280.             int c = ctl.get();
  281.             //获取线程池状态
  282.             int rs = runStateOf(c);
  283.             /**
  284.              * 只要满足下面任一条就直接返回false
  285.              * 线程池状态为STOP, TYDING 或 TERMINATD 状态
  286.              * 线程池状态为SHUTDOWN，并且firstTask != null 或者workQueue为空
  287.              */
  288.             //这里为什么不直接判断线程池状态呢？是因为有可能在线程池状态为RUNNING时已经把任务放入队列中，放入完成以后状态变为SHUTDOWN
  289.             if (rs >= SHUTDOWN && !(rs == SHUTDOWN &&
  290.                     firstTask == null && !workQueue.isEmpty()))
  291.                 return false;
  292.             for (; ; ) {
  293.                 //得到工作的线程数量
  294.                 int wc = workerCountOf(c);
  295.                 //如果工作线程数量大于理论上线程池容量；或者工作线程数大于(corePoolSize or maximumPoolSize) 直接返回false，添加失败
  296.                 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
  297.                     return false;
  298.                 //cas修改工作线程数，工作线程数+1。如果修改失败需要重新执行；成功退出循环
  299.                 if (compareAndIncrementWorkerCount(c))
  300.                     break retry;
  301.                 c = ctl.get();
  302.                 //如果线程池状态变化则需要重新执行
  303.                 if (runStateOf(c) != rs)
  304.                     continue retry;
  305.             }
  306.         }
  307.         //worker是否已经启动
  308.         boolean workerStarted = false;
  309.         //worker是否添加成功
  310.         boolean workerAdded = false;
  311.         Worker w = null;
  312.         try {
  313.             //构建worker
  314.             w = new Worker(firstTask);
  315.             //注意，这个thread不是firstTask，是从线程工厂造出来的
  316.             final Thread t = w.thread;
  317.             if (t != null) {
  318.                 //获取锁
  319.                 final ReentrantLock mainLock = this.mainLock;
  320.                 mainLock.lock();
  321.                 try {
  322.                     //获取线程池状态
  323.                     int rs = runStateOf(ctl.get());
  324.                     //如果线程池状态是RUNNING或者状态是SHUTDOWN但是队列里面还有任务
  325.                     if (rs < SHUTDOWN ||
  326.                             (rs == SHUTDOWN && firstTask == null)) {
  327.                         //如果t.isAlive()=true，说明是有问题的，都没有启动，t怎么会是活的呢。所以抛出异常。
  328.                         if (t.isAlive())
  329.                             throw new IllegalThreadStateException();
  330.                         //把worker添加到set集合
  331.                         workers.add(w);
  332.                         //记录线程池中出现的最大线程数
  333.                         int s = workers.size();
  334.                         if (s > largestPoolSize)
  335.                             largestPoolSize = s;
  336.                         //添加成功标识
  337.                         workerAdded = true;
  338.                     }
  339.                 } finally {
  340.                     //释放锁
  341.                     mainLock.unlock();
  342.                 }
  343.                 //添加成功启动线程，启动线程是调用了runWorker()
  344.                 if (workerAdded) {
  345.                     t.start();
  346.                     workerStarted = true;
  347.                 }
  348.             }
  349.         } finally {
  350.             //启动失败
  351.             if (!workerStarted)
  352.                 //启动失败需要从workers中移除当前构造的woker；工作线程数减1了；执行tryTerminate()判断是否终止线程池。
  353.                 addWorkerFailed(w);
  354.         }
  355.         return workerStarted;
  356.     }
  358.     final void runWorker(Worker w) {
  359.         //获取当前线程
  360.         Thread wt = Thread.currentThread();
  361.         //需要执行的任务
  362.         Runnable task = w.firstTask;
  363.         //置为null
  364.         w.firstTask = null;
  365.         //这里是为了把之前的state=-1设置为state=0，此时允许中断
  366.         w.unlock();
  367.         //是否异常退出循环
  368.         boolean completedAbruptly = true;
  369.         try {
  370.             //如果任务不为null或者getTask()不为null
  371.             while (task != null || (task = getTask()) != null) {
  372.                 //获取锁。这里使用锁的目的在于标识正在处理任务
  373.                 w.lock();
  374.                 //线程池=SHUWDOWN，要保证当前线程是中断状态
  375.                 //线程池!=SHUWDOWN，要保证当前线程不是中断状态
  376.                 if ((runStateAtLeast(ctl.get(), STOP) ||
  377.                         (Thread.interrupted() &&
  378.                                 runStateAtLeast(ctl.get(), STOP))) &&
  379.                         !wt.isInterrupted())
  380.                     wt.interrupt();
  381.                 try {
  382.                     //留给子类实现
  383.                     beforeExecute(wt, task);
  384.                     Throwable thrown = null;
  385.                     try {
  386.                         //执行任务
  387.                         task.run();
  388.                     } catch (RuntimeException x) {
  389.                         thrown = x;
  390.                         throw x;
  391.                     } catch (Error x) {
  392.                         thrown = x;
  393.                         throw x;
  394.                     } catch (Throwable x) {
  395.                         thrown = x;
  396.                         throw new Error(x);
  397.                     } finally {
  398.                         //子类执行
  399.                         afterExecute(task, thrown);
  400.                     }
  401.                 } finally {
  402.                     task = null;
  403.                     //更新任务完成数
  404.                     w.completedTasks++;
  405.                     w.unlock();
  406.                 }
  407.             }
  408.             //没有出现异常
  409.             completedAbruptly = false;
  410.         } finally {
  411.             //一定要注意。执行到这里说明getTask()返回null。说明当前线程池中不需要那么多线程来执行任务了，可以把多于corePoolSize数量的工作线程干掉
  412.             processWorkerExit(w, completedAbruptly);
  413.         }
  414.     }
  416.     //什么情况才会执行getTask呢？说明工作线程数已经大于核心线程数才会执行getTask()。一定要记住这一点
  417.     private Runnable getTask() {
  418.         //表示上次从阻塞队列中取任务时是否超时
  419.         boolean timedOut = false;
  420.         for (; ; ) {
  421.             int c = ctl.get();
  422.             //获取线程池状态
  423.             int rs = runStateOf(c);
  424.             //如果线程池状态rs >= SHUTDOWN并且(rs >= STOP，或者阻塞队列为空)。则将workerCount减1并返回null。
  425.             if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
  426.                 decrementWorkerCount();
  427.                 return null;
  428.             }
  429.             //获取工作线程数
  430.             int wc = workerCountOf(c);
  431.             //表示是否需要超时控制。allowCoreThreadTimeOut默认false；如果线程池数量超过核心线程数也是需要超时控制的
  432.             boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  433.             //校验工作线程数量和任务队列是否为空
  434.             if ((wc > maximumPoolSize || (timed && timedOut))
  435.                     && (wc > 1 || workQueue.isEmpty())) {
  436.                 //工作线程数量-1
  437.                 if (compareAndDecrementWorkerCount(c))
  438.                     return null;
  439.                 continue;
  440.             }
  442.             try {
  443.                 //根据是否超时从队列中获取任务
  444.                 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
  445.                 if (r != null)
  446.                     return r;
  447.                 timedOut = true;
  448.             } catch (InterruptedException retry) {
  449.                 //说明发生了中断
  450.                 timedOut = false;
  451.             }
  452.         }
  453.     }
  455.     private void processWorkerExit(Worker w, boolean completedAbruptly) {
  456.         //是否异常，如果异常工作线程数量-1
  457.         if (completedAbruptly)
  458.             decrementWorkerCount();
  460.         final ReentrantLock mainLock = this.mainLock;
  461.         mainLock.lock();
  462.         try {
  463.             //增加线程池完成任务数量
  464.             completedTaskCount += w.completedTasks;
  465.             //从worker中移除完成任务
  466.             workers.remove(w);
  467.         } finally {
  468.             mainLock.unlock();
  469.         }
  470.         //根据线程池状态进行判断是否结束线程池
  471.         tryTerminate();
  473.         int c = ctl.get();
  474.         if (runStateLessThan(c, STOP)) {
  475.             if (!completedAbruptly) {
  476.                 int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
  477.                 if (min == 0 && !workQueue.isEmpty())
  478.                     min = 1;
  479.                 if (workerCountOf(c) >= min)
  480.                     return; // replacement not needed
  481.             }
  482.             addWorker(null, false);
  483.         }
  484.     }
  486.     //根据线程池状态判断是否结束线程池
  487.     final void tryTerminate() {
  488.         for (; ; ) {
  489.             int c = ctl.get();
  490.             //RUNNING不能结束线程池
  491.             //线程池状态是TIDYING或TERMINATED说明线程池已经处于正在终止的路上
  492.             //状态为SHUTDOWN，但是任务队列不为空不能结束线程池
  493.             if (isRunning(c) ||
  494.                     runStateAtLeast(c, TIDYING) ||
  495.                     (runStateOf(c) == SHUTDOWN && !workQueue.isEmpty()))
  496.                 return;
  497.             //工作线程数量不等于0，中断一个空闲的工作线程，并返回
  498.             if (workerCountOf(c) != 0) {
  499.                 interruptIdleWorkers(ONLY_ONE);
  500.                 return;
  501.             }
  503.             final ReentrantLock mainLock = this.mainLock;
  504.             mainLock.lock();
  505.             try {
  506.                 // 设置线程池状态为TIDYING，如果设置成功，则调用terminated方法
  507.                 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
  508.                     try {
  509.                         //子类实现
  510.                         terminated();
  511.                     } finally {
  512.                         // 设置状态为TERMINATED
  513.                         ctl.set(ctlOf(TERMINATED, 0));
  514.                         termination.signalAll();
  515.                     }
  516.                     return;
  517.                 }
  518.             } finally {
  519.                 mainLock.unlock();
  520.             }
  521.         }
  522.     }
  524.     //该方法用于给空闲工作线程一个中断标识
  525.     private void interruptIdleWorkers(boolean onlyOne) {
  526.         final ReentrantLock mainLock = this.mainLock;
  527.         mainLock.lock();
  528.         try {
  529.             //遍历worker，根据onlyOne判断，如果为ture只中断一个空闲线程
  530.             for (Worker w : workers) {
  531.                 Thread t = w.thread;
  532.                 //线程没有被中断并且线程是空闲状态tryLock()判断是否空闲
  533.                 if (!t.isInterrupted() && w.tryLock()) {
  534.                     try {
  535.                         t.interrupt();
  536.                     } catch (SecurityException ignore) {
  537.                     } finally {
  538.                         w.unlock();
  539.                     }
  540.                 }
  541.                 if (onlyOne)
  542.                     break;
  543.             }
  544.         } finally {
  545.             mainLock.unlock();
  546.         }
  547.     }
  548. }
  550. //它是ThreadPoolExecutor一个内部类，此类继AbstractQueuedSynchronizer，目的在于判断线程是否空闲以及是否可以被中断
  551. private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
  552.     //工作线程空间，由线程池中所设置的线程工厂创建
  553.     final Thread thread;
  554.     //用来保存传入的任务
  555.     Runnable firstTask;
  556.     //执行完任务的总数
  557.     volatile long completedTasks;
  558.     Worker(Runnable firstTask) {
  559.         //state设置为-1是为了禁止在执行任务前对任务进行中断
  560.         setState(-1);
  561.         //提交的task
  562.         this.firstTask = firstTask;
  563.         //从线程工厂获取的线程，注意这个thread并不是用户线程
  564.         this.thread = getThreadFactory().newThread(this);
  565.     }
  567.     //实现Runnable，在run()中调用了runWorker()
  568.     public void run() {
  569.         runWorker(this);
  570.     }
  571. }

spring的ThreadPoolTaskExecutor

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

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

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

发表评论

表情：
评论列表 （有 0 条评论，284人围观）

还没有评论，来说两句吧...

相关阅读

    相关 hashMap源码解析

    源码来自jdk：1.8，和其他jdk版本可能有少许差异。 一.hashMap的实现原理     hashMap底层是一个有Node组成的数组，每个Node都有一个key

    冷不防冷不防/ 0 赞/ 528 阅读