• Android多线程学习：线程池（二）

    一、线程池运行流程

    具体执行流程如下：

    1、首先检测线程池运行状态，如果不是RUNNING，则直接拒绝，线程池要保证在RUNNING的状态下执行任务；

    2、如果workerCount < corePoolSize，则创建并启动一个线程来执行新提交的任务；

    3、如果workerCount >= corePoolSize，且线程池内的阻塞队列未满，则将任务添加到该阻塞队列中，然后空闲的核心线程会依次去缓存队列中取任务来执行；

    4、如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且线程池内的阻塞队列已满，则创建并启动一个非核心线程（临时工）来执行新提交的任务；

    5、如果workerCount >= maximumPoolSize，并且线程池内的阻塞队列已满, 则根据拒绝策略来处理该任务, 默认的处理方式是直接抛异常。

    线程池运行.png

    二、线程池状态计算

    上一节讲了线程池状态的含义及转化，这节讲状态变量计算方式：

        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
                                
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    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;

    // Packing and unpacking ctl
    private static int runStateOf(int c)     { return c & ~CAPACITY; }  //计算当前运行状态
    private static int workerCountOf(int c)  { return c & CAPACITY; }  //计算当前线程数量
    private static int ctlOf(int rs, int wc) { return rs | wc; }  //通过状态和线程数生成ctl

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    如上面代码所示：
    ThreadPoolExecutor中使用一个AtomicInteger 类来表示线程池状态及该状态下的线程数量。将一个32位的int值分为了两部分，前3位表示线程状态，后29位表示线程个数，最大可表示2^29 - 1个线程，运行线程池根本达不到这个数量。

    位数计算：
    COUNT_BITS = Integer.SIZE - 3 = 32 - 3 = 29
    容量计算：
    CAPACITY = (1 << COUNT_BITS) - 1 = 100000000000000000000000000000(30位) - 1 = 11111111111111111111111111111(29位)

    之所以这样设计是因为线程池天生处于一个并发的环下，如果分开用两个变量进行存储，就必须要通过锁进行线程安全处理，从而保证两个变量的修改具备原子性，但是这种做法对性能影响非常严重，因此将两个变量分别包装在一个变量中，最后的并发操作发生在AtomicInteger 上，而AtomicInteger 恰恰就是一个无锁原子操作类，这样既可以解决线程安全问题，又避免锁的使用，从而保证了并发性能。并且这里都使用的是位运算的方式，相比于基本运算，速度也会快很多。

    三、线程池源码阅读

    1、任务执行入口execute：

    public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
   
    int c = ctl.get();
    int num = workerCountOf(c);
    //第一步
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //第二步
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }//第三步
    else if (!addWorker(command, false))
        reject(command);
}

// firstTask 待执行的任务 core 是否要启动核心线程
private boolean addWorker(Runnable firstTask, boolean core) 

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    • 第一步，判断工作线程数如果小于corePoolSize核心线程数，直接创先新的线程addWorker，并将任务作为该线程的第一个任务(firstTask)，添加成功直接返回。如果运行线程大于等于核心线程数执行第二步；

    • 第二步，若当前线程池处于运行状态，则向阻塞队列中插入一个任务。若插入成功，则进行进行二次校验，防止一次校验通过后线程池关闭。二次校验如果线程池不是运行状态，移除任务并执行拒绝策略。如果运行状态没变，但工作线程个数是0，则新建一个非核心线程，添加个null任务，线程会去队列中获取执行任务。

      若一开始判断线程池处于非运行状态或者插入队列失败（队列任务已经满了），执行第三步；

    • 第三步，如果队列已经满了，新建一个非核心线程执行该任务，如果新建失败，可能是线程池关闭了，或者线程数量达到了maxPoolSize。

    2、新建工作线程addWorker：

    private boolean addWorker(Runnable firstTask, boolean core) {
         //第一步
        retry: //双层for循环 retry外层循环终止标识
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {         
                int wc = workerCountOf(c);
                if (wc >= CAPACITY || 
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                 //CAS：workerCount +1 
                if (compareAndIncrementWorkerCount(c))
                    //跳出外层循环 ，循环结束
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    //继续最外层循环
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 第一步，这一步主要是检查是否能新建工作线程，如果可以执行CAS操作，workCount加1。工作线程创建失败的主要原因是：
      （1）当前线程池处于关闭状态rs >= SHUTDOWN；
      （2）线程数量超过线程池设置最大值；
      （3）CAS操作失败，说明其他线程也有该操作，会在循环内重试。
            //第二步
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            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());
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                       
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                   
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //workerAdded = true; 
                if (workerAdded) {
                    // 启动线程 
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 第二步，经过第一步说明当前条件下可以创建工作线程，这一步主要是线程创建逻辑。中间修改变量workers和largestPoolSize 前需要加锁，设置成功后将锁放开，启动工作线程工作。
    private final class Worker extends AbstractQueuedSynchronizer implements Runnable  {
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    Worker类，实现了Runnable接口，并持有一个线程thread，一个初始化的任务firstTask。thread是在调用构造方法时通过ThreadFactory来创建的线程，可以用来执行任务；

    firstTask用它来保存传入的第一个任务，这个任务可以有也可以为null。如果这个值是非空的，那么线程就会在启动初期立即执行这个任务，也就对应核心线程创建时的情况；如果这个值是null，那么就需要创建一个线程去执行任务列表（workQueue）中的任务，也就是非核心线程的创建。

    3、工作线程执行

    public void run() {
    runWorker(this);
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {            
        while (task != null || (task = getTask()) != null) {
            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
            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 {
        //获取不到任务时，主动回收自己
        processWorkerExit(w, completedAbruptly);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50

    工作线程创建成功后，t.start()会启动线程执行任务，最终会进入到runWorker方法中，主要步骤如下：

    • 第一步，看下当前线程创建的时候有没有自带的任务，有的话先执行自带任务，没有的话通过getTask()去阻塞队列获取，如果获取到的task为null，则执行processWorkerExit(w, completedAbruptly)回收自己。

    • 第二步，在执行task的时候的需要加锁，如果线程池正在停止，那么要保证当前线程是中断状态，否则要保证当前线程不是中断状态。之所以要加锁是因为在工作线程回收的时候，会通过锁来判断线程是否在执行任务，如果加锁了说明在执行task则不回收。

    • 第三步， task.run()任务执行。

    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.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        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;
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35

    判断流程如下：

    • 第一步，getTask()返回 null 说明工作线程需要被回收，如下几个情况会返回null：
      （1）线程状态处于STOP，或者处于SHUTDOWN且队列中没有需要执行的task；
      （2）线程最大个数maximumPoolSize中间被修改过，导致当前线程数超过maximumPoolSize或者获取任务超时，此时如果队列中没有任务或者线程数大于1，返回null；

    • 第二步，获取timed变量值，如果核心线程设置过超时时长，或者当前线程是非核心线程，该变量会置为true，说明获取任务的时候需要加入超时判断，进行线程回收。

    • 第三步，如果线程需要限时回收执行poll()函数，如果没设置过超时，并且是核心线程，执行take()函数。poll()在指定的超时时间内没有获取到线程的task()自动超时回收，take()会一直阻塞，直到有新的task出现。

    总结，代码看多了会有点晕，前面的源码分析抽离出来大体流程图如下：
    线程池任务执行流程图.png

    4、工作线程回收

    private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly)
        decrementWorkerCount();
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {       
       completedTaskCount += w.completedTasks;       
       workers.remove(w);
    } finally {
       mainLock.unlock();
    }

    tryTerminate();
    ....
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    在执行processWorkerExit函数后，加锁修改completedTaskCount 及workers变量值，最后走到 tryTerminate()中，这个方法在任何可能导致线程池终止的动作后执行，比如减少wokerCount或SHUTDOWN状态下从队列中移除任务。

    final void tryTerminate() {
    for (;;) {
    int c = ctl.get();
    if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN &&           
               !workQueue.isEmpty()))
        return;

   if (workerCountOf(c) != 0) { // Eligible to terminate
       interruptIdleWorkers(ONLY_ONE);
       return;
   }
   final ReentrantLock mainLock = this.mainLock;
   mainLock.lock();
   try {
      if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
            try {
                // 子类重载：一些资源清理工作
                terminated();
            } finally {
              ctl.set(ctlOf(TERMINATED, 0));
              termination.signalAll();
            }
            return;
       }
   } finally {
        mainLock.unlock();
   }
// else retry on failed CAS
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29

    tryTerminate()主要分为以下几步：

    • 第一步，判断当前线程池状态，如果还处于RUNNING状态，或SHUTDOWN状态但是任务队列非空，或runState >= TIDYING 线程池已经停止了或在停止了，则直接返回；

    • 第二步，如果当前状态不属于第一步的判断，则可以将空闲的工作线程进行回收，如果工作线程个数大于0，则执行interruptIdleWorkers()。

    private void interruptIdleWorkers(boolean onlyOne) {
   final ReentrantLock mainLock = this.mainLock;
   mainLock.lock();
   try {
      for (Worker w : workers) {
         Thread t = w.thread;
         if (!t.isInterrupted() && w.tryLock()) {
              try {
                  t.interrupt();
              } catch (SecurityException ignore) {

              } finally {
                  w.unlock();
              }
       }
       if (onlyOne)
        break;
     }
  }  finally {
     mainLock.unlock();
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    从上面代码可以看到，在回收的时候会通过w.tryLock验证是否能获取到锁，如果可以获取到说明该线程没有在运行，因为runWorker中执行任务会先lock，因此保证了中断的肯定是空闲的线程。

    参考文章：
    Java线程池实现原理及其在美团业务中的实践
    全方位解析-Android中的线程池

  • 相关阅读:
    【瞎折腾】荣耀7X换脸手术，收获良多
    在虚拟机安装Hadoop
    支持向量机SVM
    LAST论文翻译
    xgplayer西瓜视频插件引用后因视频格式引起问题？
    springboot毕设项目大学生创新项目运维系统9y232（java+VUE+Mybatis+Maven+Mysql）
    RK3568开发笔记（八）：开发板烧写buildroot固件（支持hdmi屏），搭建Qt交叉编译开发环境，编译一个Demo，目标板运行Demo测试
    利用华为ENSP模拟器分析和配置中小型企业网络的综合实验（下）
    Java之JvisualVM简介
    网络安全——（黑客）自学
  • 原文地址：https://blog.csdn.net/kongqwesd12/article/details/133702274