C/C++教程

提升--17---线程池--03----ThreadPoolExecutor源码解析

本文主要是介绍提升--17---线程池--03----ThreadPoolExecutor源码解析,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

文章目录

  • ThreadPoolExecutor源码解析
    • 1、常用变量的解释
        • 1 ` ctl`,可以看做一个int类型的数字,==高3位表示线程池状态,低29位表示worker数量==
        • 4. ==线程池有5种状态==,按大小排序如下:RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
        • 在下面就是对ctl的一些操作了runStateOf取他的状态,workerCountOf计算有多少个线程正在工作,还有第8和第9个runStateLessThan、runStateAtLeast是帮助写代码的一些东西。
    • 2、构造方法
    • 3、线程池worker任务单元
        • ==这个work他本身是Runnable同时又是AQS==
    • 成员变量:
        • - final Thread thread; ------线程
        • - Runnable firstTask; -----任务
        • - volatile long completedTasks; -----任务
        • 这个work类,简单的你就可以把它当成==线程类,然后这个线程类执行的是你自己的任务==就行了。
    • 4、提交执行task的过程----execute()
      • execute()方法
        • 第一步:核心线程数不够,启动核心的;
        • 第二步:核心线程够了加队列;
        • 第三部:核心线程和队列这两个都满了,非核心线程;
    • 5、addWorker源码解析
        • addWorker就是添加线程,线程是要存到容器里
      • addWorker做两件事儿
    • 6、核心线程执行逻辑-runworker
        • runwork是真真正正启动线程之后是怎么样去执行这个任务的


ThreadPoolExecutor源码解析

1、常用变量的解释

在这里插入图片描述

// 1. `ctl`,可以看做一个int类型的数字,高3位表示线程池状态,低29位表示worker数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 2. `COUNT_BITS`,`Integer.SIZE`为32,所以`COUNT_BITS`为29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 3. `CAPACITY`,线程池允许的最大线程数。1左移29位,然后减1,即为 2^29 - 1
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
// 4. 线程池有5种状态,按大小排序如下:RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
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
// 5. `runStateOf()`,获取线程池状态,通过按位与操作,低29位将全部变成0
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 6. `workerCountOf()`,获取线程池worker数量,通过按位与操作,高3位将全部变成0
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 7. `ctlOf()`,根据线程池状态和线程池worker数量,生成ctl值
private static int ctlOf(int rs, int wc) { return rs | wc; }

/*
 * Bit field accessors that don't require unpacking ctl.
 * These depend on the bit layout and on workerCount being never negative.
 */
// 8. `runStateLessThan()`,线程池状态小于xx
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}
// 9. `runStateAtLeast()`,线程池状态大于等于xx
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}

1 ctl,可以看做一个int类型的数字,高3位表示线程池状态,低29位表示worker数量

  • 那他干嘛不用两个值,这里面肯定是自己进行了一些优化的,如果让我们自己写一定是两个值,我们线程池目前是什么状态,然后在这里面到底目前有多少个线程在运行,记录下来,只不过他把这两个值合二为一了,执行效率是会更高一些,因为这两个值都需要线程同步,所以他放在一个值里面,只要对一个线程进行线程同步就可以了,所以这里AtomicInteger在线程数量非常多,执行时间非常短的时候相对于synchronized效率会更高一些

4. 线程池有5种状态,按大小排序如下:RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED

  • RUNNING:正常运行的;
  • SHUTDOWN:调用了shutdown方法了进入了shutdown状态;
  • STOP:调用了shutdownnow马上让他停止;
  • TIDYING:调用了shutdown然后这个线程也执行完了,现在正在整理的这个过程叫TIDYING;
  • TERMINATED:整个线程全部结束;

在下面就是对ctl的一些操作了runStateOf取他的状态,workerCountOf计算有多少个线程正在工作,还有第8和第9个runStateLessThan、runStateAtLeast是帮助写代码的一些东西。

2、构造方法

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;
    // 根据传入参数`unit`和`keepAliveTime`,将存活时间转换为纳秒存到变量`keepAliveTime `中
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

3、线程池worker任务单元

这个work他本身是Runnable同时又是AQS

在这里插入图片描述

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;

    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;

    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        // 这儿是Worker的关键所在,使用了线程工厂创建了一个线程。传入的参数为当前worker
        this.thread = getThreadFactory().newThread(this);
    }

    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

    // 省略代码...
}

成员变量:

- final Thread thread; ------线程

- Runnable firstTask; -----任务

- volatile long completedTasks; -----任务

  • 这个work他本身是Runnable同时又是AQS,关于AQS这块儿你可以先忽略无所谓,因为用别的方式也能实现。本身是一个Runnable你进来的任务他又用这个Runnable给你包装了一下,为什么又要包装呢,因为它里面有好多的状态需要记录,你原来这个任务里是没有的,另外这个东西必须得在线程里运行,所以呢他用Runnable又给你包装了一次。
  • 然后这个work类里面会记录着一个成员变量,这个成员变量是thread。是哪个thread正在执行我这个对象呢,很多个线程会抢,所以这个就是为什么要用AQS的原因。另外呢,在你整个执行的过程之中你也需要加锁,不然的话你别的线程进来,要求你这个work执行其他的任务也是很有可能的
    ,这个时候也需要加锁,因此AQS是需要的。

这个work类,简单的你就可以把它当成线程类,然后这个线程类执行的是你自己的任务就行了。

4、提交执行task的过程----execute()

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * 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.
     *
     * 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.
     *
     * 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.
     */
    int c = ctl.get();
    // worker数量比核心线程数小,直接创建worker执行任务
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // worker数量超过核心线程数,任务直接进入队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 线程池状态不是RUNNING状态,说明执行过shutdown命令,需要对新加入的任务执行reject()操作。
        // 这儿为什么需要recheck,是因为任务入队列前后,线程池的状态可能会发生变化。
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 这儿为什么需要判断0值,主要是在线程池构造方法中,核心线程数允许为0
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果线程池不是运行状态,或者任务进入队列失败,则尝试创建worker执行任务。
    // 这儿有3点需要注意:
    // 1. 线程池不是运行状态时,addWorker内部会判断线程池状态
    // 2. addWorker第2个参数表示是否创建核心线程
    // 3. addWorker返回false,则说明任务执行失败,需要执行reject操作
    else if (!addWorker(command, false))
        reject(command);

execute()方法

第一步:核心线程数不够,启动核心的;

第二步:核心线程够了加队列;

第三部:核心线程和队列这两个都满了,非核心线程;

  1. execute执行任务的时候判断任务等于空抛异常,这个很简单,
  2. 接下来就是拿状态值,拿到值之后计算这个值里面的线程数,活着的那些线程数是不是小于核心线程数,如果小于addWorker添加一个线程,addWorker是比较难的一个方法,他的第二个参数指的是,是不是核心线程,所有上来之后如果核心数不够先添加核心线程,再次检查这个值。我们原来讲过这个线程里面上来之后刚开始为0,来一个任务启动一个核心线程,
  3. 第二个就是核心线程数满了之后,放到队列里。
  4. 最后核心线程满了,队列也满了,启动非核心线程。小于线程数就直接加,后面执行的逻辑就是不小于了,不小于就是超过核心线程数了直接往里扔,workQueue.offer就是把他扔进去队列里,再检查状态。在这中间可能会被改变状态值,因此需要双重检查,这个跟我们之前聊过的单例模式里面的DC是一样的逻辑。isRunning,重新又拿这个状态,拿到这个状态之后这里是要进行一个状态切换的,如果不是Running状态说明执行过shutdown命令,才会把这个Running转换成别的状态,其他情况下workerCountOf如果等于0说明里面没有线程了,没有线程我线程池正常运行就添加非核心线程。这些步骤都是通过源码可以看出来的。
  5. 如果添加work本身都不行就reject把他给拒绝掉。

5、addWorker源码解析

  • addWorker就是添加线程,线程是要存到容器里,往里头添加线程的时候务必要知道可能有好多个线程都要往里头扔,所以一定要做同步.
  • 然后呢,由于它要追求效率不会用synchronized,他会用lock或者是自旋也就增加了你代码更复杂的一个程度

addWorker就是添加线程,线程是要存到容器里

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    // 外层自旋
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 这个条件写得比较难懂,我对其进行了调整,和下面的条件等价
        // (rs > SHUTDOWN) || 
        // (rs == SHUTDOWN && firstTask != null) || 
        // (rs == SHUTDOWN && workQueue.isEmpty())
        // 1. 线程池状态大于SHUTDOWN时,直接返回false
        // 2. 线程池状态等于SHUTDOWN,且firstTask不为null,直接返回false
        // 3. 线程池状态等于SHUTDOWN,且队列为空,直接返回false
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        // 内层自旋
        for (;;) {
            int wc = workerCountOf(c);
            // worker数量超过容量,直接返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 使用CAS的方式增加worker数量。
            // 若增加成功,则直接跳出外层循环进入到第二部分
            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
        } 
    }
    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;
            // worker的添加必须是串行的,因此需要加锁
            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)) {
                    // worker已经调用过了start()方法,则不再创建worker
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // worker创建并添加到workers成功
                    workers.add(w);
                    // 更新`largestPoolSize`变量
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 启动worker线程
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // worker线程启动失败,说明线程池状态发生了变化(关闭操作被执行),需要进行shutdown相关操作
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

addWorker做两件事儿

  • 第一:count先加1;
  • 第二:才是真正的加进任务去并且start;

下面我们大致读一下,他这个里面做了两步,整个addWorker源码做了两部,上面两个for循环只是做了第一步,这个就干了一件事儿,把worker的数量加1,添加一个worker。数量在32位的那个29位里面,而且是在多线程的情况下加1,所以他进行了两个死循环干这个事儿外层死循环套内层死循环,上来先拿状态值,然后进行了一堆的判断,如果状态值不符合的话就return false,这个状态值加不进去,什么时候这个状态值不符合啊,就是大于shutdown,说明你已经shutdown了,或者去除上面这些状态之外,所有的状态都可以往里加线程。加线程又是一个死循环,首先计算当前的wc线程数是不是超过容量了,超过容量就别加了,否则用cas的方式加,如果加成功了说明第一步完成了,就retry把整个全都break掉,外层循环内层循环一下全都跳出来了,如果没加成功就get,get完了之后呢重新处理,continue retry,相当于前面在不断的试,一直试到我们把这个数加到1为止。

然后,后面才是真真正正的启动这个work,new一个work,这个work被new出来之后启动线程,这个work代表一个线程,其实这个work类里面有一个线程,加锁,是在一个容器里面,多线程的状态是一定要加锁的,锁定后检查线程池的状态,为什么要检查,因为中间可能被其他线程干掉过,看这个状态是不是shutdown了等等,如果满足往里加的条件,加进去,加完这个线程后启动开始运行,这是addWorker的一个大体逻辑。

6、核心线程执行逻辑-runworker

runwork是真真正正启动线程之后是怎么样去执行这个任务的

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 调用unlock()是为了让外部可以中断
    w.unlock(); // allow interrupts
    // 这个变量用于判断是否进入过自旋(while循环)
    boolean completedAbruptly = true;
    try {
        // 这儿是自旋
        // 1. 如果firstTask不为null,则执行firstTask;
        // 2. 如果firstTask为null,则调用getTask()从队列获取任务。
        // 3. 阻塞队列的特性就是:当队列为空时,当前线程会被阻塞等待
        while (task != null || (task = getTask()) != null) {
            // 这儿对worker进行加锁,是为了达到下面的目的
            // 1. 降低锁范围,提升性能
            // 2. 保证每个worker执行的任务是串行的
            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();
            // 执行任务,且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
            // 这两个方法在当前类里面为空实现。
            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 {
                // 帮助gc
                task = null;
                // 已完成任务数加一 
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 自旋操作被退出,说明线程池正在结束
        processWorkerExit(w, completedAbruptly);
    }
}

  • runwork是真真正正启动线程之后是怎么样去执行这个任务的,同样的,加锁。这个比较好玩的是这个work是从AbstractQueuedSynchronizer继承出来的同时实现了Runnable,说明work可以放在线程里运行,与此同时他本身就是一把锁,就可以做同步,另外,他是可以被线程执行的一个任务.

  • 为什么它本身就是一把锁啊,这个work可以认为是等着执行的一个工人,是好多个任务都可以往里面去扔内容的,也就是说会有多线程去访问这个对象的,多线程访问这个对象的时候他干脆就给自己做成了一把锁,就不要自己去定义一个lock了,所以你需要往这个work里面扔任务的时候,指定我这个线程就是你执行的这个线程的时候,好,通过work自己去lock就可以了,完全没必要再去new别的lock,所以运行work的时候就先lock住,你要run他就得lock住才能执行,不然别的线程有可能把这个work给占了, 下面又是一堆的执行,执行完了之后unlock出来,执行完了之后++ 。

这篇关于提升--17---线程池--03----ThreadPoolExecutor源码解析的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!