android开发实战 QuarkXPress 高阶函数 电力杆 editor resultMap datatable datagridview eking文件 directory devise webkit Backbonejs vue遍历 在线考试系统代码 jquery绑定change事件 jq选择子元素 bootstrap侧边栏 oracle修改字段默认值 datetimepicker赋值 quartz配置 windows杀进程命令 matlab图像滤波 二分查找python java基础教程 java实用教程 java正则表达式详解 java时间戳转日期 java怎么编译 java获取本机ip java中collection php开发教程 kafka中文教程 千元以下最好的手机 js上传图片 xs颜色 linux端口映射 tampermonkey gunzip popen函数
当前位置: 首页 > 学习教程  > 编程语言

深入理解线程池——细致入微的讲源码。

2021/2/13 17:20:20 文章标签: 测试文章如有侵权请发送至邮箱809451989@qq.com投诉后文章立即删除

在上一篇博文《图解线程池原理》中,大体上介绍了线程池的工作原理。 这一篇从源码层面,细致剖析,文章会很长。 如果上篇文章内容没吸收,先看上篇,先易后难嘛。 一、示例代码 public static void main(String[] args…

在上一篇博文《图解线程池原理》中,大体上介绍了线程池的工作原理。

这一篇从源码层面,细致剖析,文章会很长。

如果上篇文章内容没吸收,先看上篇,先易后难嘛。

一、示例代码


    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        for(int i =1; i <= 10; i++){
            int index = i;
            pool.execute(new Runnable() {
                @Override
                public void run() {
                    String currentThreadName = Thread.currentThread().getName();
                    log.info("第{}次任务结束,执行者:{}", index, currentThreadName);
                }
            });
        }
        pool.shutdown();
        System.out.println("All thread is over");
    }
    

类图关系, Executor 是最顶级接口,只有一个 execute() 方法,

最终大数的实现方法,在 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;
    }

这里有7 个参数,功能如下

  1. corePoolSize 核心线程数
  2. maximumPoolSize 最大线程数
  3. keepAliveTime 线程空闲时,最大存活时间
  4. unit 存活时间的单位
  5. workQueue 阻塞队列(存放任务)
  6. threadFactory 线程工厂(生产线程用)
  7. handler 拒绝策略

二、线程控制参数

先看这个,这个COUNT_BITS ,就是数字 29,下面会用到。

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

CAPACITY 最大线程数,(二进制数就是 3 个 0, 29个1)

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

在这里插入图片描述
线程池运行状态


    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;
    

如图,int 值的高三位代表线程运行状态,后 29 位记录工作线程数量
在这里插入图片描述
线程池的五种状态,转化关系如图所示
在这里插入图片描述


	private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
	private static int ctlOf(int rs, int wc) { return rs | wc; }
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    private static int workerCountOf(int c)  { return c & CAPACITY; }

ctl 是原子递增的一个数,这个数高3位,表示线程池的运行状态,后29位是记录运行的线程数。

在这里插入图片描述
每增加一个线程,ctl 增加 1, 每销毁1个线程,ctl 减小1

runStateOf 就是获取线程池的运行状态,即 ctl 的高三位

在这里插入图片描述
图片上,两个数字,进行与运算,得到的 ctl 值的高三位,低29位一定都是0,

现在不明白,后文用到的时候就明白了。

同样的,workerCountOf 是获取正在运行的线程数量,即 ctl 的低29位。

在这里插入图片描述
图片上的两个数字,进行与运算,得到的是 ctl 的低29位的值,(高 3 位一定是0啦)

图中这种状态,线程池是 RUNNING 状态(看高3位),运行的线程数量是 13(低29位的值)

Doug Lea 真的是很牛,用一个数字,即控制了线程池的运行状态,又记录了线程的数量。

三、线程池的运行


	public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) { // 1、工作线程数量小于核心线程数,创建线程
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) { // 2、将任务放入阻塞队列
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false)) // 3、队满时,继续创建线程,直至工作线程数量达到最大值。
            reject(command); // 4、执行拒绝策略
    }

先明确一点:

传入进来的任务,要么被 addWorker() 方法接收了,要么被放到阻塞队列里,要么被拒绝策略给收了。

拿上一篇的例子来说,病人进来,要么被医生叫走了,要么呆在大厅里,等着被医生叫走,要么被拒绝策略收走了。

第一步:

workerCountOf(c) < corePoolSize 这个判断,前半部分是获取运行的线程数,

这个不明白,往上翻看,看看那个图。

如果该判断成立,走 addWorker() 方法。

第二步:

工作线程数量达到了corePoolSize,那准备往阻塞队列中放,当然在在线程池的处于 RUNNING 状态。

// 方法很容易懂,ctl 是负数就可以了
    private static boolean isRunning(int c) {
        return c < SHUTDOWN;
    }

如果将任务成功放入阻塞队列了,紧接着做了一个判断 ! isRunning(recheck) && remove(command)

这是线程池不是运行状态,就从阻塞队列中删除任务。

仔细想想,前后判断了两次运行状态,这里是做了极端情况,并发处理,不明说了。

第二个判断 workerCountOf(recheck) == 0,这个是干啥的?开始我也没想清楚。

后来明白了,这个是所有医生都休息了,然后队列有任务,派个医生穿上防护服,去等着病人。

这个具体后面再分析,大概知道下,不明白没关系,后面再解释。

第三步:

任务放到阻塞队列失败了,那就创建工作线程。

第四步:

前面几步都失败了,执行拒绝策略。

总体上这个方法就讲完了,没有锁、没有CAS,不怕有并发么?

锁控制是在addWorker() 方法中,这是尽可能减小锁的粒度,提高性能。

然后讲重头戏,addWorker() ,先打下预防针,不太好理解。

  • addWorker() 方法解析
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
            }
        }

        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();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

先说这一段


            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
                

看到这一大串,正常都会晕,我也晕。即某些情况下,addWorker() 方法直接返回 false

我看了很多次,这段代码与下面的同一个意思。


            if (rs > SHUTDOWN) return false;
            if (rs = SHUTDOWN && firstTask != null) return false;
            if (rs = SHUTDOWN && workQueue.isEmpty()) return false;
            

也就是说,线程池是 RUNNING 可以执行 addWorker() 方法,

线程池是 SHUTDOWN 且队列不为空,且是空任务时,可以执行 addWorker() 方法,

其他情况一律不得执行。为什么第二个条件,那么复杂,后面讲到再说。

看下 for 循环


	retry:
	for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c); // 线程池运行状态
            for (;;) {
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize)) // 控制线程数量
                    return false;
                if (compareAndIncrementWorkerCount(c)) // 线程数量+1
                    break retry; // 跳出外层 for 循环,执行下面的操作
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs) 
                    continue retry; // 如果线程运行状态变化,从外层循环开始,重新执行,否则从内层循环开始,重新执行。
            }
        }


    private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }
    

这个双层 for 循环,就是做了一件事,在符合的条件时,ctl 增加1。

这里 break retrycontinue retry 的作用,都写了注释,语法不熟悉的,得问度娘。

然后再看下半段


		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 {
                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) { // 前面说过addWorker 执行的条件,RUNNING, 或者 SHUTDOWN 且 空任务
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w); // 核心,将worker放入集合中
                        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;

这段代码中 new Worker(firstTask); 是一个重点要说的。 这是 Worker 这个类的继承关系图
在这里插入图片描述
看下这个类的简化代码


    private final class Worker extends AbstractQueuedSynchronizer
        implements Runnable
    {
        private static final long serialVersionUID = 6138294804551838833L;
        final Thread thread; // 线程
        Runnable firstTask; // 任务
        volatile long completedTasks; // 已完成的任务数
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        public void run() {
            runWorker(this);
        }
    }

Worker 继承了 AbstractQueuedSynchronizer,就有了可重入锁的功能,
实现了 Runnable,就重写了 run() 方法。

new Worker(firstTask) 时,将任务赋值了,创建了线程,并且加了锁。

这里还要说明下,ThreadPoolExecutor 类中的一个全局变量 workers

 	/**
     * Set containing all worker threads in pool. Accessed only when
     * holding mainLock.
     */
    private final HashSet<Worker> workers = new HashSet<Worker>();
    

源码的注释写的很清楚,workers 是存放 worker 对象的集合,

且只有拿到锁的worker对象,才会被放到该集合中。

后半段的核心逻辑是创立线程,即 new 一个 Worker对象,把对象放到集合中。然后启动线程。

至此,execute() 方法讲的差不多了,除了拒绝策略没说。

四、循环执行任务

  • addWorker 方法解析

addWorker 方法中会调用 Thread 类中的 start() 方法。然后JVM 会在合适的时间调用 run() 方法。


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

runWorker 的本质是执行 task.run(),而 task 获取 是将 阻塞队列取空为止。

顺便说一句,第一次执行时,task 是 worker 在初始化时的那个任务,之后是从阻塞队列中取任务。

这种设计,实现了一个线程执行了多个任务。

说细节:

w.unlock(); // allow interrupts 这行代码,是不是和我一样,刚开始觉得很奇怪。

怎么就好端端的来一个 释放锁呢? 这是因为在 new Worker对象时,加了锁

        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker:执行runWorker 不会被中断
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

关于 AQS 的加锁与解锁,这里不详细解释。不理解的可以看我之前的博文《ReentrtantLock 分析》

while (task != null || (task = getTask()) != null) 这是循环取任务,等下详细说。

下面这一段看着都会头晕,谁看谁头晕。


                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                    

源码注释说的很明确,

当线程池状态,大于等于 STOP 时,保证工作线程都有中断标志。

当线程池状态,小于STOP时,保证工作线程都没有中断标志。

这里详细解释下: 先看前半部分

runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))



    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }

runStateAtLeast(ctl.get(), STOP) = A

那 伪代码就是 A || (Thread.interrupted() && A)

情况1:A 为 true,那后面不会执行了,整体为true

情况2:A 为false, 执行 Thread.interrupted() ,即清除线程的中断标志,
(Thread.interrupted() && A) 这个条件是 false, 整体结果为false。

前半部分说完了,说后半部分&& !wt.isInterrupted()

上文 情况2时,不会执行后半部分,即判断结束了。A 为 false,清除线程中断标志。

这也就实现了,线程池状态小于 STOP,工作线程保证没有中断标志。

情况1时,执行 wt.isInterrupted(), 如果有中断,返回true,取反后,整体是false。

如果没中断,返回false,取反后,整体是 true,那会执行 wt.interrupt(),即打个中断标志。

不管怎样,最终线程会有中断标志。

也就实现了,线程池状态大于 等于STOP,工作线程保证有中断标志。

有时候,代码不好理解,真不是读代码的人笨,也不是写代码的人,写的不好

.
.

  • processWorkerExit 方法解析

继续说 runWorker 方法,当阻塞队列中取出的任务是null时,会跳出 while 循环,执行 processWorkerExit()

这个方法是在 finally 中执行的,正常情况下,是从阻塞队列中取任务,取到的是null,此时 completedAbruptly是 false。

当然非正常情况,某个环节抛出了异常,执行了这个方法,此时completedAbruptly 是 true.


private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            workers.remove(w); // 删除线程
        } finally {
            mainLock.unlock();
        }

        tryTerminate(); // 优雅的尝试关闭线程池

        int c = ctl.get();
        if (runStateLessThan(c, STOP)) {
            if (!completedAbruptly) {
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            addWorker(null, false);
        }
    }

如果是出了异常,才执行这个方法的,那工作线程的数量减一,这个好理解。

不论是否出异常,都会在加锁的情况下,删除 worker,也就是删除了线程。

当然,在删除线程之前,会统计个线程完成了多少个任务。

tryTerminate() 这个方法,等到 shutDown() 时再详细说。这里只要知道,这个方法是有可能关闭线程池的。

继续说这个方法:processWorkerExit ,在最后,当线程池状态是 RUNNING、SHUTDOWN 时,有可能会再次调用 addWorker 方法,只是传入的任务是null。

首先是 completedAbruptly 为 true 时,调用 addWorker。这个好理解,上个方法异常退出了,那再起一个线程,继续消费任务。

其次是completedAbruptly 为 false 时,即真的是从阻塞队列中取的任务是null,工作线程的数量小于 min, 这时会调用 addWorker。

顺便说一句,当工作线程数量,小于或等于 corePoolSize,正常不会进到这个方法的。为啥,等会儿说这个。

int min = allowCoreThreadTimeOut ? 0 : corePoolSize 这里 allowCoreThreadTimeOut 默认是false。这里 min 是可以等于corePoolSize的。这种情况的出现,想不明白一定不是作者写错了。

corePoolSize 是可以重置的, setCorePoolSize(int corePoolSize) 。特定时间的重置 corePoolSize ,这里的代码判断就有必要了。

总之这个方法会销毁线程,也会在必要的情况下创建线程,保证工作线程不小于corePoolSize 。
.
.

  • getTask() 方法解析

现在退回来,说 runWorker 方法中的这一段 while (task != null || (task = getTask()) != null)

getTask() 方法返回 null 时,runWorker 会执行结束,进入 processWorkerExit 方法中,销毁线程。

我曾经错误的以为,当阻塞队列中取不到任务了,进入processWorkerExit 这个方法中。

销毁工作线程,但同时维护corePoolSize ,还会创建线程。

那不就有可能,前面创建,后面销毁,循环不停啦!

后来我发现我错了,原因就在 getTadk() 的源码中。

    /**
     * Performs blocking or timed wait for a task, depending on
     * current configuration settings, or returns null if this worker
     * must exit because of any of:
     * 1. There are more than maximumPoolSize workers (due to
     *    a call to setMaximumPoolSize).
     * 2. The pool is stopped.
     * 3. The pool is shutdown and the queue is empty.
     * 4. This worker timed out waiting for a task, and timed-out
     *    workers are subject to termination (that is,
     *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
     *    both before and after the timed wait, and if the queue is
     *    non-empty, this worker is not the last thread in the pool.
     *
     * @return task, or null if the worker must exit, in which case
     *         workerCount is decremented
     */
    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);

            // Are workers subject to culling?
            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;
            }
        }
    }

英文好的,可以先看下源码的注释

这又是一个无限循环,先看这段线程池运行状态的判断,看着会晕

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

这个的意思就是

rs = SHUTDOWN && workQueue.isEmpty() 执行 大括号里的代码

rs > SHUTDOWN 时,执行大括号里的代码

也就是说,这两种情况下,返回null,进入 processWorkerExit 方法销毁线程了。

其它情况(rs = RUNNING,或者 rs = SHUTDOWN 并且 阻塞队列不为空),要尝试返回任务。

接着看这一行

	boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

timed 这个参数,你可以理解为,是否支持超时机制

false 不支持,从后文的代码可看出,取任务时,调用 workQueue.take()阻塞,代码就停到这里了,一直等到,其它线程往队列中放了任务,阻塞被唤醒,继续执行。

true 支持, 取任务时,调用 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS),即在一定时间内,取不到任务,直接返回 null。

allowCoreThreadTimeOut 参数的默认值是 false, 那咱们就只讨论 wc > corePoolSize。

当 wc > corePoolSize 时,timed 为 true,即支持超时,一定时间取不到任务,返回 null。

结合前面所讲,可以总结这么两句话:

当工作线程大于 corePoolSize,取任务时,没有任务,操作会超时,返回null,进入下一次循环。

当工作线程 不大于 corePoolSize,取任务时,没有任务,会被阻塞,即代码停止执行,直到有新的任务进来。

再看下面这一段 ,可以理解为 是否具备销毁线程的条件


            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            

timedOut 这个参数,可以理解为:是否经历过取任务超时,false 没有经历过, true 经历过。

条件1: wc > maximumPoolSize || (timed && timedOut)

条件2: wc > 1 || workQueue.isEmpty()

wc > maximumPoolSize 条件1就满足了。什么时候出现呢? 当重置 maximumPoolSize时会出现这种情况。

那条件1,可以用文字描述为:要么工作线程数大于 最大线程数,要么是 经历过取任务超时(支持超时机制,才有可能经历过取任务超时),才可能去销毁线程。

条件1满足的情况下,才会判断条件2

wc > 1 这个说明当前工作线程,至少是 2 个,可以销毁 1 个。

workQueue.isEmpty() 这个说明阻塞队列中没有任务了,可以放心销毁线程。


    private boolean compareAndDecrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }
    

如果 compareAndDecrementWorkerCount 执行成功了,那就直接返回null。

如果执行失败了,那就从头开始,重新判断。这个不难理解,并发控制的真好

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }

前面说的理解好了,这里就容易理解了。

支持超时等待,就调用 非阻塞poll 方法,不支持的调用 阻塞 方法 take()

创建线程池时,指定的参数 最大线程存活时间 keepAliveTime, 就是在这里发挥功效的。

至此,getTask() 方法讲完了,笼统的概括:

当工作线程 小于等于 corePoolSize,取任务会调用 阻塞方法 take()

即队列中没有任务会被阻塞,也就是说,线程不会被销毁。

当工作线程 大于 corePoolSize, 取任务会调用 非阻塞的 poll() 方法,

即队列中没有任务时,会超时,在下个循环中,返回 null 进入销毁线程的流程。
.
.

五、线程池的关闭

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess(); // 检查有没有权限关闭线程池
            advanceRunState(SHUTDOWN); // 更改线程池的状态
            interruptIdleWorkers(); // 中断线程
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

这里的代码很简洁,来一个一个分析

  • 1、checkShutdownAccess() 方法解析

    private void checkShutdownAccess() {
        SecurityManager security = System.getSecurityManager();
        if (security != null) {
            security.checkPermission(shutdownPerm);
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                for (Worker w : workers)
                    security.checkAccess(w.thread);
            } finally {
                mainLock.unlock();
            }
        }
    }

这个方法逻辑很清楚,先检查有没有权限关闭线程池,

如果有,再检查是否可以中断每个工作线程,

没有相关权限,会抛出异常。
.

  • 2、advanceRunState() 方法解析

这个方法,最终是把线程池的运行状态,设置为一个大于等于0 的状态,即大于等于 SHUTDOWN


    private void advanceRunState(int targetState) {
        for (;;) {
            int c = ctl.get();
            if (runStateAtLeast(c, targetState) ||
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }
    
    private static boolean runStateAtLeast(int c, int s) {
        return c >= s;
    }

入参是 SHUTDOWN,如果线程池的运行状态 大于等于 SHUTDOWN,直接跳出。

否则,执行 ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))) 这一段

workerCountOf(c) 是工作线程的数量,执行这它的时候,线程池的状是 RUNNING,

workerCountOf(c) 是大于等于0的。所以 ctlOf(targetState, workerCountOf(c))) 计算出来也是大于0的。

ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) 执行的结果就是,线程池的状态设置为大于等于 0。

.

  • 3、interruptIdleWorkers() 方法解析

这个方法,是中断空闲线程。

    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }


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

首先 mainLock.lock() 表明,整个过程是加锁进行的,且是拿到主锁才执行。

!t.isInterrupted() && w.tryLock() 这个条件,

若 t.isInterrupted() 返回 true,即线程本身有中断标志,取反后是false,后面的就不再执行了。

顺便说一句,在 解析 runWorker() 方法时,分析过,

当线程池状态,大于等于 STOP 时,保证工作线程都有中断标志。

当线程池状态,小于STOP时,保证工作线程都没有中断标志。

工作线程没有中断标志的情况下,会执行 w.tryLock() 方法。

有没有疑问,这是什么意思呀?

前面说过,Worker 对象 继承了 AQS,也就是说,它可以实现加锁。

回头可以再看下 runWorker() 方法,在执行这个方法之前,不会被中断,

在执行任务时,是加锁的,也不会被中断,

在执行 getTask() 方法时,是可以被中断的。


		public boolean tryLock()  { return tryAcquire(1); }

        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        

这段加锁的代码很容易理解,接着往下说。

在执行任务的线程不会被中断,这也就实现了,即使线程池关闭了,正在执行任务的线程,依然会把任务给执行完。

拿到锁之后,会执行 t.interrupt(), 即给线程打上中断标志。

本例中,传入的参数是 false,那会对所有的空闲线程打上中断标志。
.

那下一个问题,线程被打上中断标志后,有啥影响呢?


结论:在执行 getTask() 方法时,会抛出异常,跳出这个方法,进入processWorkerExit() 销毁线程。

而且此时,线程池的状态若是等于SHUTDOWN,就会会调用 addWorker(null, false) 。

前面分析 addWorker() 方法时,我们知道只有两种情况下,会创建一个新的 Worker。

不记得的话,翻回去看下。 结合现在的场景,只有队列不为空的时候,分创建新的 Worker,

若队列为空,不会创建新的Worker。

想想,队列不为空,说明还有任务存在,销毁了一个线程,再创建一个,也正常。

由此,把前面分析过的几个方法都串了起来。

如果说,上面的分析你理解了,那下面这句话你会有新的认识:

当线程池是SHUTDOWN 状态时,队列不为空时,是可以提交空任务。

当线程池是STOP 状态时,不可以提交任务。
.

咱们还有一个问题没有说,中断的线程怎么就抛出异常了?

在 getTadk() 方法中, 取任务的那行代码

       Runnable r = timed ?
           workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
           workQueue.take();

在 BlockQueue这个接口中,定义了这两个方法,是会抛出异常的。


    E poll(long timeout, TimeUnit unit)
        throws InterruptedException;

    E take() throws InterruptedException;
    

以 LinkedBlockingQueue的实现为例子。详细信息可以看了之前的博文《LinkedBlockingQueue源码图解》


    public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly(); // 线程有中断,直接抛出异常
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

下面是 ReentrantLocklockInterruptibly() 方法的实现


    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
    
    public final void acquireInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (!tryAcquire(arg))
            doAcquireInterruptibly(arg);
    }

至此为止,interruptIdleWorkers() 这个方法解析完了,总结下大概就是

线程池状态大于等于SHUTDOWN 时,会调用该方法,将所有空闲的线程打个中断标志。

被打中断标志的线程,会在 getTask() 方法中抛出异常,从而在后一个方法中销毁线程。

若销毁线程后,阻塞队列中还有任务,还是可以新建 Worker,继续消费队列中的任务。
.

  • 4、tryTerminate() 方法解析

onShutdown() 这个方法,在 ThreadPoolExecutor 类中是一个空方法,这里就不讲了。

tryTerminate() 这个方法,在 processWorkerExit() 里也调用过,当时没讲,放在这里解析


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

先看这一段,判断线程池状态的

       if (isRunning(c) ||
           runStateAtLeast(c, TIDYING) ||
           (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
           return;

线程池处于运行状态,不处理,这个好理解。

线程池状态 大于等于TIDYING,不处理。这个说明已经在成功执行过 tryTerminate 方法了。

线程池状态 是SHUTDOWN 且 阻塞队列不为空,不处理,因为有任务,所以线程池不关闭。

除此之外,就两咱状态需要处理了:

线程池状态是STOP,可以关闭线程池。
线程池状态是SHUTDOWN 且阻塞队列为空。


        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE); // 中断一个空闲线程。
            return;
        }
        

这段代码的意思我懂,当工作线程大于0时, 中断一个空闲的线程。

interruptIdleWorkers 这个方法,前面讲过了,只是参数变成了true,只中断一个空闲线程。

shutdown() 这个方法中,所有空闲线程都被中断了。简化的考虑,这段没用。

但是在 processWorkerExit() 方法中,调用 tryTerminate() 是有意义的。

它可以让阻塞的线程,从 getTask() 方法中,跳出来,进入processWorkerExit() 被销毁掉。

同时会触发再次中断一个空闲线程,使其从getTask() 方法中,跳出来,进入processWorkerExit() 被销毁掉。

这就形成了一个循环,把空闲的线程一个一个处理掉了。

那么回过头来想想, shutdown() 调用 interruptIdleWorkers(true),也是有意义的。

因为正在执行任务的线程,执行完任务,是可能变成空闲线程的。(大概知道就行了,不深入解释)

再往后,获取主锁,修改线程池的状态。

先是设置为 TIDYING状态,再执行 terminated(),最后线程池状态设置为 TERMINATED。

其中 terminated()ThreadPoolExecutor 这个类中,是一个空方法,不用讲了。

最后 termination.signalAll(), 这个是唤醒阻塞的线程,这个方法本文的方法中未曾涉及,不讲了。

至此线程池源码解析,完毕。

六、总结

本文从源码层面,详细分析了,线程池的创建、运行、关闭。

对应的就是 execute(), addWorker(), runWorker(), shutDown() 这几个方法。

同时与简单梳理了,这几个方法之间相互关联的地方。

上篇博文《图解线程池原理》,从大方向上介绍了线程池的原理 ,本篇深入源码,详细剖析了这几个就去。

关于线程池的四种拒绝策略,以后单独写一篇,不要本文中分析的。

还有一些方法,比如 shutdownNow()、awaitTermination() 等 不再详细分析,文章已经够长啦!

能把本文所讲的东西说出来,面试官基本该满意了吧。


本文链接: http://www.dtmao.cc/news_show_700122.shtml

附件下载

相关教程

    暂无相关的数据...

共有条评论 网友评论

验证码: 看不清楚?