十三、线程池ThreadPoolExecutor源码解析

在学习了CAS、原子类、AQS、各种锁、并发队列之后我们开始学习线程池，既是对前面的巩固也是对技术更进一步探索，可以说大部分的框架都离不开线程池，
所以理解它对我们后期的学习有非常大的帮助

介绍

线程池主要解决两个问题：

  • 当执行大量并发任务时，线程池能提供较好的性能，不需要每次使用new来创建线程对象，减少开销
  • 线程池实现了对线程的管理和资源限制，以及一些统计数据

在concurrent包中我们可以使用Executors工具类创建线程池，工具包中有不同的实现，根据需要，返回不同的线程池实例。
ThreadPoolExecutor是其中最基础也是用的最多的一种，所以我们从它开始入手

类图构成

线程池.PNG

从类图中可以看到，它有一个原子变量ctl，它是用来记录线程池状态和线程个数的，有点类似读写锁中的state

  • private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  • private static final int COUNT_BITS = Integer.SIZE - 3; // 计数位为int类型位数-3（下面我们假设是32位），也就是低29位
  • private static final int CAPACITY = (1 << COUNT_BITS) - 1; // 容量是29个1
  • private static final int RUNNING = -1 << COUNT_BITS; // 高三位11100000000000000000000000000000
  • private static final int SHUTDOWN = 0 << COUNT_BITS; // 000
  • private static final int STOP = 1 << COUNT_BITS; // 001
  • private static final int TIDYING = 2 << COUNT_BITS; // 010
  • private static final int TERMINATED = 3 << COUNT_BITS; // 011
    private static int runStateOf(int c)     { return c & ~CAPACITY; } // 获取高三位
    private static int workerCountOf(int c)  { return c & CAPACITY; }  // 获取低29位
    private static int ctlOf(int rs, int wc) { return rs | wc; }    // 获取ctl值

如果看过读写锁部分，就应该觉得很熟悉的操作了，都是通过位运算来获取状态，其高三位代表状态：

  • RUNNING：处理阻塞队列里的任务，并且接受新任务
  • SHUTDOWN：处理阻塞队列里的任务，但不接受新任务 调用shutdown()方法时
  • STOP：中断任务，且不处理阻塞队列里的任务，并且不接受新任务 显式调用shutdownNow()
  • TIDYING：表示在所有任务执行完后（包括阻塞队列）执行terminated方法,当线程池为空
  • TERMINATED 终止状态3

其他核心参数：

  • BlockingQueue<Runnable> workQueue; 用于保存等待执行的任务的阻塞队列，也就是我们之前讲到过的那些并发队列
  • ReentrantLock mainLock 独占锁，用来控制worder的原子性，比如新增worker
  • HashSet<Worker> workers 线程池执行任务的对象，也就是真正工作的线程
  • int corePoolSize; 核心线程个数
  • int maximumPoolSize; 线程池最大线程数量
  • RejectedExecutionHandler handler 饱和策略，也就是队列满了，并且线程个数达到最大值采取的处理方式，如抛出异常、丢弃但不抛出异常
  • keepAliveTime 当线程数量超过了corePoolSize指定的线程数，并且空闲线程空闲的时间达到当前参数指定的时间时该线程就会被销毁，如果调用过allowCoreThreadTimeOut(boolean value)方法允许核心线程过期，那么该策略针对核心线程也是生效的
  • threadFactory: 创建线程的工厂，如果未指定则使用默认的线程工厂

方法源码分析

execute

线程池使用的核心方法是execute，该方法的作用是提交任务到线程池中执行。

  • execute 
    public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
    //三种情况，第一种是当前线程个数小于corePollSize，开启新的新的线程
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))// 有可能线程数达到核心线程限制，则失败
            return;
        c = ctl.get();
    }
    // 第二种情况，如果状态是Running，则添加任务到阻塞队列。
    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);    // 如果是
    }
    // 第三种情况，如果队列吗，则新增线程，新增失败再执行拒绝策略
    else if (!addWorker(command, false))
        reject(command);
}

总结上面的调度策略：

  1. 如果线程池中的线程数小于corePoolSize，那么每来一个任务都会创建一个新的线程
  2. 若当前执行的任务达到了corePoolSize指定的线程数时，也即所有的核心线程都在执行任务时，此时来的新任务会保存在workQueue指定的任务队列中；也就是第二种情况
  3. 当所有的核心线程都在执行任务，并且任务队列中存满了任务，此时若新来了任务，那么线程池将会创建新线程执行任务；也就是第三种情况
  4. 如果队列满了，并且线程数达到了maximumPoolSize，则reject ，也就是第三种情况但新增失败。（所以任务的最大数是queueSize+maximumPoolSize）

由上面代码可知，主要逻辑集中在addworker上

  • addWorker(Runnable firstTask, boolean core) 
    private boolean addWorker(Runnable firstTask, boolean core) {
    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;
            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
        }
    }
    // 主要分两部分，前半部分使用cas来更新工作线程数量，后半部分使用加锁来创建线程。
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);  // work线程
        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();
            }
            if (workerAdded) {
                t.start();      // 启动线程
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}
    addWorker方法中由于没有做同步，因此有很多判断线程状态的逻辑，我们注意到一般判断都是以shutdown为分界线，shutdown以上自不必说，肯定不能再创建线程了，只要注意如果线程池为shutdown状态时如何处理就行，只有当入参firstTask为null，且队列不为空时才会增加工作线程，其他不增加。 那么重点就在于什么时候会入参为null呢？我们先往下看，再回来解答这个问题。

Worker执行

要了解Worker这个类，我就可以从它的官方注释说起

  1. 为什么要继承AQS，也就是为什么要锁？

    This protects against interrupts that are intended to wake up a worker thread waiting for a task from instead interrupting a task being run.
    翻译过来就是，为了中断正在等待任务的线程，而不是中断正在运行的线程。所以，实现锁，主要是为了运行中的线程不被中断（加锁就不会）。

  2. 为什么不使用ReentrantLock而是自己实现呢？

    because we do not want worker tasks to be able to reacquire the lock when they invoke pool control methods like setCorePoolSize.
    翻译过来就是，不想让在调用线程池方法时，重入获取锁。（会调用interruptIdleWorkers，tryLock会重入，来中断线程）

  • 构造函数 
    Worker(Runnable firstTask) {
        setState(-1); // 防止执行runWorker之前的中断
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
  • run 这个是现成的方法，实际里面执行的是runWorker(this) 
    public void run() {
        runWorker(this);
    }

结合上文中说到再addWorker成功后会启动线程，也就是执行这里的runWorker(this)方法。

  • runWorker 
    final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts  state设置为0可以相应中断了。 
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            // 到这说明还有任务，开始执行任务
            w.lock();  // 防止被shutdown
            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);// 清理函数
    }
}
    函数的作用就是不断的循环，去getTask取任务，如果有任务则执行任务，在没有任务后，线程会执行清理并退出函数
  • getTasK 获取任务 
    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())) {
            // 到这说明没有任务了，则将工作线程worker减1
            decrementWorkerCount(); 
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 如果设置了超时时间，或者线程数超出核心线程数则为true。都使用超时时间

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            // 到这说明，线程数量已经足够多，或者已经超时，则返回null，并减少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;
        }
    }
}
    上面这个函数主要关注超时部分，两种情况，一种是核心线程需要超时，另一种就是线程数超过核心线程。
  • processWorkerExit 
    private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted,表示意外结束，则表示在run()发生了异常，则需要数量-1 ，如果不是，则不需要，因为在getTask()中-1了
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();// 释放线程，需要加锁。
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);  // 从workers中移除
    } finally {
        mainLock.unlock();
    }

    tryTerminate();// 尝试终止线程池，因为有可能现在已经没有任务了
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) { // running状态或shutdown状态，则还要增加一个线程
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // 如果大于核心线程数则退出
        }
        // 也就是说小于核心线程数才会增加
        addWorker(null, false);
    }
}
    注意之前的问题，什么时候addWorker参数为null时，shutdown状态也是可以添加的，也就是要去完成队列里的任务。

关闭线程池

  • shutdown
    调用shutdown方法后，线程池不再接受任务，但工作队列中的任务还是会被执行

     public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();// 权限检查
        advanceRunState(SHUTDOWN);  // 设置状态
        interruptIdleWorkers();     // 中断等待任务的线程（1）
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}   
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）这里可能会疑惑，怎么中断的？回到runWorker()函数中，我们发现能执行到这里说明拿到锁了，那么中断可能在task != null || (task = getTask()) != nul这条语句中。其实就是在getTask()中

    往getTask()函数下面看，找到，worQueue.poll或者workQueue.take();这两个函数都是获取任务队列中的下个任务，而且都是加锁且响应中断。所以我们中断的作用是是在这。
    当然肯定还会问，只是结束这一次的循环获取任务而已，并不会中断啊。确实没错，它仅仅是结束这一次的获取。但是，如果这是队列处在SHUTDOWN并且队列为空时，那么就会退出，并且移除线程，从而达到中断清理线程的目的。那如果队列中还存在呢，这又回到我们之前说的，shutdown状态，仅仅是不再接受任务，但还是会处理队列中的任务。所以还是会继续获取任务，只有等任务处理完毕，才开始一个一个退出

    2）在shutdown执行完，仍有任务，队列不会退出，那最后是如何清理的呢。

    答案是之前讲过的processWorkerExit(w, completedAbruptly); 在没有任务之后，线程会自动退出从workers中移除，然后执行tryTerminate()

  • tryTerminate()

     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); //中断workers中的一个
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 到这说明已经没有线程了 workerCountOf(c) == 0，则设置状态为TERMINATED
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();  // 唤醒所有等待terminate的线程
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

  • shutdownNow
    shutdownNow和shutdown函数不一样，它会结束所有的线程，

        public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}
private void interruptWorkers() {   // 不是interruptIdleWorkers
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            w.interruptIfStarted(); // 无需获取锁，也就是执行中的任务也会被中断。
    } finally {
        mainLock.unlock();
    }
}

总结

  1. 创建线程池
    使用Executors工具类创建，根据最大线程数、核心线程数、超时空闲时间来创建。

  2. 添加任务

    • 如果线程数小于核心线程数，则直接添加线程
    • 如果线程数达到核心线程数，则将任务添加到任务阻塞队列
    • 如果线程数达到核心线程数，且任务队列满了，则创建新线程
    • 如果线程数达到最大线程数，且任务队列满了，则拒绝任务

  3. 执行任务

    • 在创建线程时获取任务或者循环从任务队列中获取任务，并执行
    • 在线程数小于核心线程数时，会一直阻塞获取任务

  4. 回收线程

    • 如果一直没有获取到线程且超过核心线程则退出
    • 调用shutdown后，会中断等待任务的线程，如果没有任务，则全部退出。如果有任务，则核心线程数内的线程会继续去等待任务，所有任务完成后，退出
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,189评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,577评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,857评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,703评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,705评论 5赞 366
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,620评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,995评论 3赞 396
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,656评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,898评论 1赞 298
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,639评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,720评论 1赞 330
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,395评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,982评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,953评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,195评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,907评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,472评论 2赞 342