zoukankan      html  css  js  c++  java
  • [多线程] Java线程池应用及原理分析(JDK1.8)

    一  线程池优点

      1、线程在创建和销毁时是非常耗费资源的,使用线程池可以减少创建和销毁线程的次数,每个工作线程都可以重复使用。

      2、可以根据系统的承受能力,调整线程池中工作线程的数量,防止因为消耗过多内存导致服务器崩溃。

      对操作系统来说,创建一个线程的代价是十分昂贵的, 需要给它分配内存、列入调度,同时在线程切换的时候还要执行内存换页,CPU 的缓存被清空。切换回来的时候还要重新从内存中读取信息,破坏了数据的局部性。

      关于时间,创建线程使用是直接向系统申请资源的,这里调用系统函数进行分配资源的耗时不好说。

      关于资源,Java线程的线程栈所占用的内存是在Java堆外的,所以是不受java程序控制的,只受系统资源限制,默认一个线程的线程栈大小是1M(当让这个可以通过设置-Xss属性设置,但是要注意栈溢出问题),但是,如果每个用户请求都新建线程的话,1024个用户光线程就占用了1个G的内存,如果系统比较大的话,一下子系统资源就不够用了,最后程序就崩溃了。

      PS:同样的道理在java程序中也不要随意开启新的线程,特别是高频业务尽量使用线程池,不然很容易导致内存不足,程序崩溃的问题。

    二  线程池创建

      java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类,因此如果要透彻地了解Java中的线程池,必须先了解这个类。下面我们来看一下ThreadPoolExecutor类的具体实现源码。所有的ThreadPoolExecutor构造函数最终都是调用的七个参数的构造函数:

      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.acc = System.getSecurityManager() == null ?  null : AccessController.getContext();
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
      }

      其中参数分别为:   

        corePoolSize:线程池核心线程数量。
                在创建了线程池后,默认情况下,线程池中并没有任何线程,而是等待有任务到来才创建线程去执行任务,
                除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法:
    prestartAllCoreThreads():预先创建所有核心线程;prestartCoreThread():预先创建一个核心线程
                当提交一个任务时,线程池创建一个新线程执行任务,直到当前线程数等于corePoolSize;如果当前线程数为corePoolSize,继续提交的任务被保存到阻塞队列中,等待被执行。

        maximumPoolSize:线程池最大线程数量
                  如果当前阻塞队列满了,且继续提交任务,则创建新的线程执行任务,前提是当前线程数小于maximumPoolSize。

        keepAliverTime:当活跃线程数大于核心线程数时,空闲的多余线程最大存活时间。
                 默认情况下,只有当线程池中的线程数大于corePoolSize时,keepAliveTime才会起作用,直到线程池中的线程数不大于corePoolSize.
                 如果调用了allowCoreThreadTimeOut(boolean)方法,在线程池中的线程数不大于corePoolSize时,keepAliveTime参数也会起作用,直到线程池中的线程数为0

        unit:存活时间的单位

        workQueue:存放任务的队列    

        t
    hreadFactory:线程工厂,一般使用默认即可

        handler:超出线程范围和队列容量的任务的处理程序(拒接策略)

     任务处理流程

      提交一个任务到线程池中,即执行execute()方法后,线程池的处理流程如下:

      1、判断线程池里的核心线程是否都在执行任务,如果不是(核心线程空闲或者还有核心线程没有被创建)则创建一个新的核心线程来执行任务。如果核心线程都在执行任务,则进入下个流程。

      2、线程池判断阻塞队列是否已满,如果阻塞队列没有满,则将新提交的任务存储在这个阻塞队列里。如果阻塞队列满了,则进入下个流程。

      3、判断线程池里的线程数量是否达到最大线程数,如果没有,则创建一个新的工作线程来执行任务。如果已经达到最大线程数,则交给拒绝策略来处理这个任务。

        

     任务缓存队列及排队策略

      在前面我们多次提到了任务缓存队列,即workQueue,它用来存放等待执行的任务。

      workQueue的类型为BlockingQueue<Runnable>,通常可以取下面三种类型:

      1)ArrayBlockingQueue:基于数组的先进先出队列,此队列创建时必须指定大小;

      2)LinkedBlockingQueue:基于链表的先进先出队列,如果创建时没有指定此队列大小,则默认为Integer.MAX_VALUE,即为无界队列。因此,如果线程池中线程数达到了corePoolSize,且始终没有空闲线程(任务提交的平均速度快于被处理的速度),任务缓存队列可能出现无限制的增长。

      3)synchronousQueue:这个队列比较特殊,它不会保存提交的任务,而是将直接新建一个线程来执行新来的任务,每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态。因此,如果线程池中始终没有空闲线程(任务提交的平均速度快于被处理的速度),可能出现无限制的线程增长。

     任务拒绝策略

      当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略,通常有以下四种策略:

      ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
      ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。
      ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
      ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务

     线程池的关闭

      ThreadPoolExecutor提供了两个方法,用于线程池的关闭,分别是shutdown()和shutdownNow(),其中:

        shutdown():不会立即终止线程池,而是要等所有任务缓存队列中的任务都执行完后才终止,但再也不会接受新的任务

        shutdownNow():立即终止线程池,并尝试打断正在执行的任务,并且清空任务缓存队列,返回尚未执行的任务

     线程池实现原理

    1  ThreadPoolExecutor类的属性

      // 线程池的控制状态,用高3位来表示线程池的运行状态,低29位来表示线程池中工作线程的数量
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        //值为29,用来表示偏移量
        private static final int COUNT_BITS = Integer.SIZE - 3;
        //线程池的最大容量,其值的二进制为:00011111111111111111111111111111(29个1)
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
        // 线程池的运行状态,总共有5个状态,用高3位来表示
        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;
    
        //任务缓存队列,用来存放等待执行的任务
        private final BlockingQueue<Runnable> workQueue;
    
        //全局锁,对线程池状态等属性修改时需要使用这个锁
        private final ReentrantLock mainLock = new ReentrantLock();
    
        //线程池中工作线程的集合,访问和修改需要持有全局锁
        private final HashSet<Worker> workers = new HashSet<Worker>();
    
        // 终止条件
        private final Condition termination = mainLock.newCondition();
    
        //线程池中曾经出现过的最大线程数
        private int largestPoolSize;
        
        //已完成任务的数量
        private long completedTaskCount;
    
        //线程工厂
        private volatile ThreadFactory threadFactory;
    
        //任务拒绝策略
        private volatile RejectedExecutionHandler handler;
    
        //线程存活时间
        private volatile long keepAliveTime;
    
        //是否允许核心线程超时
        private volatile boolean allowCoreThreadTimeOut;
    
        //核心池大小,若allowCoreThreadTimeOut被设置,核心线程全部空闲超时被回收的情况下会为0
        private volatile int corePoolSize;
    
        //最大池大小,不得超过CAPACITY
        private volatile int maximumPoolSize;
    
        //默认的任务拒绝策略
        private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
    
        private static final RuntimePermission shutdownPerm = new RuntimePermission("modifyThread");
    
        private final AccessControlContext acc;

    2  线程池重要成员变量

    (1)AtomicInteger ctl

      private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

      ctl代表了ThreadPoolExecutor中的控制状态,它是一个复合类型的成员变量,是一个原子整数,借助高低位包装了两个概念:

           workerCount:线程池中当前活动的线程数量,占据ctl的低29位;在线程池成员变量里用COUNT_BITS表示。

           runState:线程池运行状态,占据ctl的高3位,有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED五种状态。

    (2)COUNT_BITS和CAPACITY

        由于线程池通过workerCount表示当前活动的线程数量,它占据ctl的低29位,这样,每当活跃线程数增加或减少时,ctl直接做相应数目的增减即可,十分方便。而ThreadPoolExecutor中COUNT_BITS就代表了workerCount所占的位数,定义如下:

      private static final int COUNT_BITS = Integer.SIZE - 3;

      在Java中,一个int占据32位,因此Integer.SIZE = 32,因此COUNT_BITS的大小就是29。另外,既然workerCount代表了线程池中当前活动的线程数量,那么它肯定有个上下限阈值,下限很明显就是0,上限是ThreadPoolExecutor中CAPACITY值,它是ThreadPoolExecutor中理论上的最大活跃线程数,其定义如下:

      private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

      运算过程为1左移29位,也就是00000000 00000000 00000000 00000001 --> 001 0000 00000000 00000000 00000000,再减去1的话,就是 000 11111 11111111 11111111 11111111即CAPACITY的值,前三位代表线程池运行状态runState,所以这里workerCount的理论最大值就应该是29个1,即536870911

      由于workerCount作为其中一个部分复合在AtomicInteger ctl的低29位中,那么ThreadPoolExecutor理应提供从AtomicInteger ctl中解析出workerCount即当前活跃的线程数量的方法:

      private static int workerCountOf(int c)  { 
        return c & CAPACITY;
      }

      计算逻辑很简单,入参 c 是 ctl 的值(ctl.get()),即高3位为线程池运行状态runState,低29位为线程池中当前活动的线程数量workerCount,将其与CAPACITY进行与操作&,也就是与00011111 11111111 11111111 11111111进行与操作,c的前三位通过与000进行与操作,无论c前三位为何值,最终都会变成000,也就是舍弃前三位的值,而c的低29位与29个1进行与操作,c的低29位还是会保持原值,这样就从AtomicInteger ctl中解析出了workerCount的值。

    (3)runState 

      线程池定义了几个static final变量表示线程池的各个状态:

      // runState is stored in the high-order bits
        private static final int RUNNING    = -1 << COUNT_BITS;
        //-1在Java底层是由32个1表示的,左移29位的话,即111 00000 00000000 00000000 00000000,也就是低29位全部为0,高3位全部为1的话,表示RUNNING状态,即-536870912.
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
      //0在Java底层是由32个0表示的,无论左移多少位,还是32个0,即000 00000 00000000 00000000 00000000,也就是低29位全部为0,高3位全部为0的话,表示SHUTDOWN状态,即0;
        private static final int STOP       =  1 << COUNT_BITS;
      //1在Java底层是由前面的31个0和1个1组成的,左移29位的话,即001 00000 00000000 00000000 00000000,也就是低29位全部为0,高3位为001的话,表示STOP状态,即536870912;
        private static final int TIDYING    =  2 << COUNT_BITS;
      //2在Java底层是由前面的30个0和1个10组成的,左移29位的话,即010 00000 00000000 00000000 00000000,也就是低29位全部为0,高3位为010的话,表示TIDYING状态,即1073741824;
        private static final int TERMINATED =  3 << COUNT_BITS;
      //2在Java底层是由前面的30个0和1个11组成的,左移29位的话,即011 00000 00000000 00000000 00000000,也就是低29位全部为0,高3位为011的话,表示TERMINATED状态,即1610612736;

      (1)RUNNING:当创建线程池后,初始时,线程池处于RUNNING状态,线程可以接受新任务,并处理队列任务。

      (2)SHUTDOWN:如果调用了shutdown()方法,则线程池处于SHUTDOWN状态,此时线程池不能够接受新的任务,它会等待所有任务执行完毕;

      (3)STOP:如果调用了shutdownNow()方法,则线程池处于STOP状态,此时线程池不能接受新的任务,并且会去尝试终止正在执行的任务;

      (4)TIDYING:所有的任务已结束,workerCount为0,线程过渡到TIDYING状态,将会执行terminated()钩子方法

      (5)TERMINATED:当线程池处于SHUTDOWN或STOP状态,并且所有工作线程已经销毁,任务缓存队列已经清空或执行结束后,terminated()方法已经完成,线程池被设置为TERMINATED状态。

       由上面我们可以得知,运行状态的值按照RUNNING-->SHUTDOWN-->STOP-->TIDYING-->TERMINATED顺序值是递增的。

       线程池获取运行状态代码如下:

      private static int runStateOf(int c) { 
        return c & ~CAPACITY;
      }

       ~是按位取反的意思,CAPACITY表示的是高位的3个0,和低位的29个1,而~CAPACITY则表示高位的3个1,低位的29个0即111 00000 00000000 00000000 00000000,然后再与入参c执行按位与操作,即高3位保持原样,低29位全部设置为0,也就获取了线程池的运行状态runState。

      最后看下 ctl 的 ctlOf 定义如下:  

      private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
      private static int ctlOf(int rs, int wc) { return rs | wc; }

      即RUNNING状态值和0做或运算:传入的rs表示线程池运行状态runState,其是高3位有值,低29位全部为0的int,而wc则代表线程池中有效线程的数量workerCount,其为高3位全部为0,而低29位有值的int,将runState和workerCount做或操作 | 处理,即用runState的高3位,workerCount的低29位填充的数字,而默认传入的runState、workerCount分别为RUNNING和0。

    2  线程池execute()方法执行原理

      在execute()方法中主要有三步操作:

      /*
       * 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.
       * 1、如果少于corePoolSize数量的线程正在运行,尝试利用给定的Runnable实例command创建一个新的线程作为它的第一个任务来执行。
       * addWorker()方法的调用会对线程池运行状态runState、线程数量workerCount进行原子性检测,返回值为启动新线程结果。
       *
       * 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.
       * 2、如果一个任务可以成功地进入队列,然后我们还需要再次检查(即双重检查)自从进入这个方法后,我们是否应该添加一个线程(因为自从上一次检查以来可能存在死亡情况),
       * 所以我们重新检查状态,如果有必要的话,即线程池已停止,回滚之前的入队操作,或者在没有线程时启动一个新线程。
       *
       * 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.
       * 3、如果我们不能入列一个任务,那么我们尝试添加一个新线程。
       * 如果添加失败,我们知道线程池可能已被关闭或者数量饱和,所以我们会拒绝这个任务。
       */
         public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            // 获取ctl的值 c,ctl 通过高低位表示线程状态和活跃的线程数量
            int c = ctl.get();
            
            // 通过workerCountOf(c)方法获取ctl的低29位数据,即活跃的线程数量。如果c中有效线程数目小于corePoolSize大小,尝试创建新的core线程处理任务command:
         // 调用addWorker(command,true)方法创建新的线程执行command任务 // 线程数的判断利用corePoolSize作为边界约束条件 // 方法返回值是标志添加worker是否成功的标志位,ture表示成功,false表示失败, // 如果为true,则直接返回,否则重新获取ctl的值c if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; // 添加work线程失败则再次获取ctl的值 c = ctl.get(); } // 根据 c 判断当前线程池的状态是否为RUNNING状态,即既可以接受新任务,又会处理队列任务的状态,如果isRunning返回true则执行workQueue.offer(command)方法 // 通过offer()方法,尝试将commond添加到队列workQueue中,BlockingQueue的offer()方法表示将参数对象加到BlockingQueue里,如果BlockingQueue可以容纳,则返回true,否则返回false,不会抛出异常 if (isRunning(c) && workQueue.offer(command)) { // 如果当前线程池处于RUNNING状态,且workQueue能够容纳command,并添加成功的话, // 再次获取ctl的值recheck, int recheck = ctl.get(); // 如果当前线程池的状态不是RUNNING,并且从队列workQueue移除command成功的话, // 调用reject()方法拒绝任务command, if (!isRunning(recheck) && remove(command)) reject(command); // 否则如果当前工作线程woker数目为0,尝试添加新的worker线程,但是不携带任务,即从阻塞队列里获取任务执行 else if (workerCountOf(recheck) == 0) addWorker(null, false); }
         // 如果加入队列失败,则会创建新的线程执行任务,这里addWoker入参为false,表示不是核心线程
    // 如果尝试添加新的worker线程处理任务command失败, // 调用reject()方法拒绝任务command,线程数的判断利用maximumPoolSize作为边界约束条件 else if (!addWorker(command, false)) reject(command); }

       其中addWorker方法传入当前任务,并尝试创建一个线程去执行任务:

      private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
            for (;;) { //自旋
                int c = ctl.get(); //获取ctl值
                int rs = runStateOf(c); //获取线程池运行状态
    
                // 如果rs >= SHUTDOWN 表示当前线程池处于SHUTDOWN、STOP、TINYING或者TERMINAL状态,即不能接受新的任务
                if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
                    return false;
    
                for (;;) {
                    int wc = workerCountOf(c);//获取活跃线程数量
                    if (wc >= CAPACITY || //活跃线程数已经达到了最大容量
                        wc >= (core ? corePoolSize : maximumPoolSize))//或者创建的是核心线程,但是活跃线程数已经达到了最大核心线程数,或者创建的非核心线程,但是活跃线程数已经达到了最大线程数,则返回false
                        return false;
                    if (compareAndIncrementWorkerCount(c))//原子操作增加线程数量:ctl.compareAndSetr(expect,expect + 1);
                        break retry;//操作成功则跳出循环
                    c = ctl.get();  //否则重新获取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);//创建一个Worker对象,入参为要执行的任务 final Thread t = w.thread;//创建线程 if (t != null) { 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());//获取线程池运行状态             //如果线程池是运行状态或者线程是SHUTDOWN状态但是firstTask是null if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // 检查线程是否存活 throw new 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 { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }

       其中w = new Worker(firstTask);final Thread t = w.thread;表示创建一个新的线程,其中worker的构造函数如下:

      Worker(Runnable firstTask) {
         setState(-1); // inhibit interrupts until runWorker
         this.firstTask = firstTask;
         this.thread = getThreadFactory().newThread(this);
       }

       在创建Thread时把当前worker作为runnable入参传给了当前线程,因此在执行t.start()方法是,会执行当前worker的run()方法:

      public void run() {
         runWorker(this);
       }
      final void runWorker(Worker w) {
            //获取当前线程
            Thread wt = Thread.currentThread();
            //获取w的firstTask
            Runnable task = w.firstTask;
            //设置w的firstTask为null
            w.firstTask = null;
            // 释放锁,设置AQS的state为0,允许中断
            w.unlock();
            //用于标识线程是否异常终止,finally中processWorkerExit()方法会有不同逻辑
            boolean completedAbruptly = true;
            try {
                //循环调用getTask()获取任务,不断从任务缓存队列获取任务并执行
                while (task != null || (task = getTask()) != null) {
                    //进入循环内部,代表已经获取到可执行的任务,则对worker对象加锁,保证线程在执行任务过程中不会被中断
                    w.lock();
                    if ((runStateAtLeast(ctl.get(), STOP) ||  //若线程池状态大于等于STOP,那么意味着该线程要中断
                         (Thread.interrupted() &&      //线程被中断
                          runStateAtLeast(ctl.get(), STOP))) &&  //且是因为线程池内部状态变化而被中断
                        !wt.isInterrupted())           //确保该线程未被中断
                        //发出中断请求
                        wt.interrupt();
                    try {
                        //开始执行任务前的Hook方法
                        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 {
                            //执行任务后的Hook方法
                            afterExecute(task, thrown);
                        }
                    } finally {
                        //置空task,准备通过getTask()获取下一个任务
                        task = null;
                        //completedTasks递增
                        w.completedTasks++;
                        //释放掉worker持有的独占锁
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally {
                //到这里,线程执行结束,需要执行结束线程的一些清理工作
                //线程执行结束可能有两种情况:
                //1.getTask()返回null,也就是说,这个worker的使命结束了,线程执行结束
                //2.任务执行过程中发生了异常
                //第一种情况,getTask()返回null,那么getTask()中会将workerCount递减
                //第二种情况,workerCount没有进行处理,这个递减操作会在processWorkerExit()中处理
                processWorkerExit(w, completedAbruptly);
            }
        }

    runWorker()方法都做了哪些工作:

      1.运行第一个任务firstTask之后,循环调用getTask()方法获取任务,不断从任务缓存队列获取任务并执行;

      2.获取到任务之后就对worker对象加锁,保证线程在执行任务的过程中不会被中断,任务执行完会释放锁;

      3.在执行任务的前后,可以根据业务场景重写beforeExecute()和afterExecute()等Hook方法;

      4.执行通过getTask()方法获取到的任务

      5.线程执行结束后,调用processWorkerExit()方法执行结束线程的一些清理工作

    在执行worker的方法中,首先会执行firstTask,firstTask执行完之后会执行getTask()通过自旋从阻塞队列中获取任务:

      private Runnable getTask() {
            boolean timedOut = false; // Did the last poll() time out?
    
            for (;;) {//自旋
                int c = ctl.get();//获取ctl
                int rs = runStateOf(c);//获取线程池状态
    
                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                    decrementWorkerCount();
                    return null;
                }
    
                int wc = workerCountOf(c);//获取线程数量
    
                // Are workers subject to culling?
           //是否设置了核心线程超时或者线程数量是否大于核心线程数,则timed为true,表示需要超时
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            //如果线程数量大于最大线程数或者设置了核心线程超时且需要超时,则直接返回null,让线程自然销毁
                if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
                    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;
                }
            }
        }

    接下来总结一下getTask()方法会在哪些情况下返回:

      1.线程池处于RUNNING状态,阻塞队列不为空,返回成功获取的task对象

      2.线程池处于SHUTDOWN状态,阻塞队列不为空,返回成功获取的task对象

      3.线程池状态大于等于STOP,返回null,回收线程

      4.线程池处于SHUTDOWN状态,并且阻塞队列为空,返回null,回收线程

      5.worker数量大于maximumPoolSize,返回null,回收线程

      6.线程空闲时间超时,返回null,回收线程

    processWorkerExit()的实现

    processWorkerExit()方法负责执行结束线程的一些清理工作,下面分析一下其实现。

      private void processWorkerExit(Worker w, boolean completedAbruptly) {
            //如果用户任务执行过程中发生了异常,则需要递减workerCount
            if (completedAbruptly)
                decrementWorkerCount();
    
            final ReentrantLock mainLock = this.mainLock;
            //获取全局锁
            mainLock.lock();
            try {
                //将worker完成任务的数量累加到总的完成任务数中
                completedTaskCount += w.completedTasks;
                //从workers集合中移除该worker
                workers.remove(w);
            } finally {
                //释放锁
                mainLock.unlock();
            }
            //尝试终止线程池
            tryTerminate();
            //获取线程池控制状态
            int c = ctl.get();
            if (runStateLessThan(c, STOP)) {  //线程池运行状态小于STOP
                if (!completedAbruptly) {  //如果用户任务执行过程中发生了异常,则直接调用addWorker()方法创建线程
                    //是否允许核心线程超时
                    int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                    //允许核心超时并且workQueue阻塞队列不为空,那线程池中至少有一个工作线程
                    if (min == 0 && ! workQueue.isEmpty())
                        min = 1;
                    //如果工作线程数量workerCount大于等于核心池大小corePoolSize,
                    //或者允许核心超时并且workQueue阻塞队列不为空时,线程池中至少有一个工作线程,直接返回
                    if (workerCountOf(c) >= min)
                        return;
                    //若不满足上述条件,则调用addWorker()方法创建线程
                }
                //创建新的线程取代当前线程
                addWorker(null, false);
            }
        }

    八  静态方法创建线程池

      在java doc中,并不提倡我们直接使用ThreadPoolExecutor,而是使用Executors类中提供的几个静态方法来创建线程池:

      Executors.newCachedThreadPool();        //创建一个缓冲池,缓冲池容量大小为Integer.MAX_VALUE
      Executors.newSingleThreadExecutor();   //创建容量为1的缓冲池
      Executors.newFixedThreadPool(int);    //创建固定容量大小的缓冲池

       下面是这三个静态方法的具体实现;

      public static ExecutorService newFixedThreadPool(int nThreads) {
          return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
      }
      public static ExecutorService newSingleThreadExecutor() {
          return new FinalizableDelegatedExecutorService
              (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
      }
      public static ExecutorService newCachedThreadPool() {
          return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
      }

      从它们的具体实现来看,它们实际上也是调用了ThreadPoolExecutor,只不过参数都已配置好了。

      newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的,它使用的LinkedBlockingQueue,即如果活跃线程数已经达到了核心线程数,则后续任务都放在阻塞队列里,不再创建新的任务。它和SingleThreadExecutor类似,唯一的区别就是核心线程数不同,并且由于使用的是LinkedBlockingQueue,在资源有限的时候容易引起OOM异常。

      newSingleThreadExecutor将corePoolSize和maximumPoolSize都设置为1,也使用的LinkedBlockingQueue;即只创建一个线程,后续任务都放在阻塞队列里,不再创建新的任务。因为LinkedBlockingQueue是长度为Integer.MAX_VALUE的队列,可以认为是无界队列,因此往队列中可以插入无限多的任务,在资源有限的时候容易引起OOM异常,同时因为无界队列,maximumPoolSize和keepAliveTime参数将无效,压根就不会创建非核心线程。

      newCachedThreadPool将corePoolSize设置为0,将maximumPoolSize设置为Integer.MAX_VALUE,使用的SynchronousQueue,也就是说来了任务就创建线程运行,当线程空闲超过60秒,就销毁线程。因为Integer.MAX_VALUE非常大,可以认为是可以无限创建线程的,在资源有限的情况下容易引起OOM异常。

      阿里巴巴开发手册并发编程这块有一条:线程池不允许使用Executors去创建,而是通过ThreadPoolExecutor的方式,通过上面可以看出,使用默认的方式创建线程池,很容易出现OOM的情况。

     如何合理配置线程池的大小

      一般需要根据任务的类型来配置线程池大小:

      如果是CPU密集型任务(计算密集型),就需要尽量压榨CPU,参考值可以设为 NCPU+1

      如果是IO密集型任务(IO操作、网络操作),由于线程阻塞时不耗费CPU资源,因此可以把线程数设置大一些,参考值可以设置为2*NCPU

      当然,这只是一个参考值,具体的设置还需要根据实际情况进行调整,比如可以先将线程池大小设置为参考值,再观察任务运行情况和系统负载、资源利用率来进行适当调整。

    十  如果线程执行抛出异常,线程池会怎么处理这个线程

    1  示例代码

    public class ExecutorsTest {
    
        public static void main(String[] args) {
            ThreadPoolTaskExecutor executorService = buildThreadPoolTaskExecutor();
            executorService.execute(() -> sayHi("execute"));
            executorService.submit(() -> sayHi("submit"));
        }
    
        private static void sayHi(String name) {
            String printStr = "【thread-name:" + Thread.currentThread().getName() + ",执行方式:" + name + "】";
            System.out.println(printStr);
            throw new RuntimeException(printStr + ",我异常啦!哈哈哈!");
        }
    
        private static ThreadPoolTaskExecutor buildThreadPoolTaskExecutor() {
            ThreadPoolTaskExecutor executorService = new ThreadPoolTaskExecutor();
            executorService.setThreadNamePrefix("(线程池)-");
            executorService.setCorePoolSize(5);
            executorService.setMaxPoolSize(10);
            executorService.setQueueCapacity(1000);
            executorService.setKeepAliveSeconds(30);
            executorService.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            executorService.initialize();
            return executorService;
        }
    }

    执行结果如下:

    2020-06-29 20:43:03.603 [main] INFO  org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService
    【thread-name:(线程池)-1,执行方式:execute】
    Exception in thread "(线程池)-1" java.lang.RuntimeException: 【thread-name:(线程池)-1,执行方式:execute】,我异常啦!哈哈哈!
        at com.dh.yjt.SpringBootDemo.test.DuoXianCheng.ExecutorsTest.sayHi(ExecutorsTest.java:18)
        at com.dh.yjt.SpringBootDemo.test.DuoXianCheng.ExecutorsTest.lambda$main$0(ExecutorsTest.java:11)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
    【thread-name:(线程池)-3,执行方式:submit】

      从执行结果我们看出

    当执行方式是execute时,可以看到堆栈异常的输出。

    当执行方式是submit时,堆栈异常没有输出。

      那么我们怎么拿到submit执行方式的堆栈异常呢,通过如下方式:

      public static void main(String[] args) throws InterruptedException {
            ThreadPoolTaskExecutor executorService = buildThreadPoolTaskExecutor();
            executorService.execute(() -> sayHi("execute"));
            TimeUnit.MILLISECONDS.sleep(10);
            System.out.println("====================");
            Future f = executorService.submit(() -> sayHi("submit"));
            try {
                f.get();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

     执行结果如下:

    2020-06-29 20:50:58.308 [main] INFO  org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService
    【thread-name:(线程池)-1,执行方式:execute】
    Exception in thread "(线程池)-1" java.lang.RuntimeException: 【thread-name:(线程池)-1,执行方式:execute】,我异常啦!哈哈哈!
        at com.dh.yjt.SpringBootDemo.test.DuoXianCheng.ExecutorsTest.sayHi(ExecutorsTest.java:28)
        at com.dh.yjt.SpringBootDemo.test.DuoXianCheng.ExecutorsTest.lambda$main$0(ExecutorsTest.java:14)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
    ====================
    java.util.concurrent.ExecutionException: java.lang.RuntimeException: 【thread-name:(线程池)-3,执行方式:submit】,我异常啦!哈哈哈!
        at java.util.concurrent.FutureTask.report(FutureTask.java:122)
        at java.util.concurrent.FutureTask.get(FutureTask.java:192)
        at com.dh.yjt.SpringBootDemo.test.DuoXianCheng.ExecutorsTest.main(ExecutorsTest.java:19)
    Caused by: java.lang.RuntimeException: 【thread-name:(线程池)-3,执行方式:submit】,我异常啦!哈哈哈!
    【thread-name:(线程池)-3,执行方式:submit】
        at com.dh.yjt.SpringBootDemo.test.DuoXianCheng.ExecutorsTest.sayHi(ExecutorsTest.java:28)
        at com.dh.yjt.SpringBootDemo.test.DuoXianCheng.ExecutorsTest.lambda$main$1(ExecutorsTest.java:17)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
        at java.util.concurrent.FutureTask.run(FutureTask.java)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

      submit方法执行时,返回结果封装在future中,如果调用future.get()方法则必须进行异常捕获,从而可以抛出(打印)堆栈异常。

    2  线程抛异常不会影响其他线程的执行

      从上面的示例可以看出,一个线程抛出了异常,不会影响其他线程的任务执行。

      但是细心的可以看出线程的名称从1直接到了3,而没有了线程2.

      

       

       

       

    3  总结

    当一个线程池里面的线程异常后:

      1. 当执行方式是execute时,可以看到堆栈异常的输出。

      2. 当执行方式是submit时,堆栈异常没有输出。但是调用Future.get()方法时,可以捕获到异常。

      3. 不会影响线程池里面其他线程的正常执行。

      4. 线程池会把这个线程移除掉,并创建一个新的线程放到线程池中。

    十一  Tomcat线程池实现原理

      从 三  任务处理流程,可以看出在JDK线程池中,任务执行是在队列无法放入新的任务时,在判断当前运行线程是否达到最大线程数,如果未达到,则创建一个新的线程执行任务。但是在Tomcat中,它里面的线程池的运行过程是先把最大线程数用完,然后再提交任务到队列里面去的。

      可参考:https://mp.weixin.qq.com/s/n28nH8xL6dTlw_vCi0wI6w

    参考

      1、Java并发编程:线程池的使用  https://www.cnblogs.com/dolphin0520/p/3932921.html

      2、ThreadPoolExecutor源码分析(一):重要成员变量  https://blog.csdn.net/lipeng_bigdata/article/details/51232266

                                 https://blog.csdn.net/lipeng_bigdata/article/details/51243348

      3、与运算(&)、或运算(|)等:https://blog.csdn.net/xiaopihaierletian/article/details/78162863

      https://mp.weixin.qq.com/s?__biz=MzIxNTQ4MzE1NA==&mid=2247483741&idx=1&sn=238fc933c3b9b19ab1754b23283ac6fd&chksm=9796d720a0e15e364f4105c29de606e7329760a41607136a722dc97bb177f9362aeacd92f762&scene=21#wechat_redirect

  • 相关阅读:
    常吃二十种降血脂食物,三高不再缠身
    员工能力要从“人海战术”转向“精兵强将”
    企业家必备的4项核心能力
    优秀管理者在哪些方面超乎常人
    高血压 降压方法
    教育视频
    吉他和弦 学习
    spoj 375 QTREE
    hihocoder #1260 : String Problem I
    codeforces 282E. Sausage Maximization Trie
  • 原文地址:https://www.cnblogs.com/aiqiqi/p/10688426.html
Copyright © 2011-2022 走看看