【高并发】通过源码深度分析线程池中Worker线程的执行流程-阿里云开发者社区

开发者社区> 冰河001> 正文

【高并发】通过源码深度分析线程池中Worker线程的执行流程

简介: 在《高并发之——通过ThreadPoolExecutor类的源码深度解析线程池执行任务的核心流程》一文中我们深度分析了线程池执行任务的核心流程,在ThreadPoolExecutor类的addWorker(Runnable, boolean)方法中,使用CAS安全的更新线程的数量之后,接下来就是创建新的Worker线程执行任务,所以,我们先来分析下Worker类的源码。
+关注继续查看

大家好,我是冰河~~

在《高并发之——通过ThreadPoolExecutor类的源码深度解析线程池执行任务的核心流程》一文中我们深度分析了线程池执行任务的核心流程,在ThreadPoolExecutor类的addWorker(Runnable, boolean)方法中,使用CAS安全的更新线程的数量之后,接下来就是创建新的Worker线程执行任务,所以,我们先来分析下Worker类的源码。

Worker类分析

Worker类从类的结构上来看,继承了AQS(AbstractQueuedSynchronizer类)并实现了Runnable接口。本质上,Worker类既是一个同步组件,也是一个执行任务的线程。接下来,我们看下Worker类的源码,如下所示。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    private static final long serialVersionUID = 6138294804551838833L;
    //执行任务的线程类
    final Thread thread;
    //初始化执行的任务,第一次执行的任务
    Runnable firstTask;
    //完成任务的计数
    volatile long completedTasks;
    //Worker类的构造方法,初始化任务并调用线程工厂创建执行任务的线程
    Worker(Runnable firstTask) {
        setState(-1); 
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
    //重写Runnable接口的run()方法
    public void run() {
        //调用ThreadPoolExecutor类的runWorker(Worker)方法
        runWorker(this);
    }

    //检测是否是否获取到锁
    //state=0表示未获取到锁
    //state=1表示已获取到锁
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }
    
    //使用AQS设置线程状态
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    //尝试释放锁
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

在Worker类的构造方法中,可以看出,首先将同步状态state设置为-1,设置为-1是为了防止runWorker方法运行之前被中断。这是因为如果其他线程调用线程池的shutdownNow()方法时,如果Worker类中的state状态的值大于0,则会中断线程,如果state状态的值为-1,则不会中断线程。

Worker类实现了Runnable接口,需要重写run方法,而Worker的run方法本质上调用的是ThreadPoolExecutor类的runWorker方法,在runWorker方法中,会首先调用unlock方法,该方法会将state置为0,所以这个时候调用shutDownNow方法就会中断当前线程,而这个时候已经进入了runWork方法,就不会在还没有执行runWorker方法的时候就中断线程。

注意:大家需要重点理解Worker类的实现。

Worker类中调用了ThreadPoolExecutor类的runWorker(Worker)方法。接下来,我们一起看下ThreadPoolExecutor类的runWorker(Worker)方法的实现。

runWorker(Worker)方法

首先,我们看下RunWorker(Worker)方法的源码,如下所示。

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    //释放锁,将state设置为0,允许中断任务的执行
    w.unlock();
    boolean completedAbruptly = true;
    try {
        //如果任务不为空,或者从任务队列中获取的任务不为空,则执行while循环
        while (task != null || (task = getTask()) != null) {
            //如果任务不为空,则获取Worker工作线程的独占锁
            w.lock();
            //如果线程已经停止,或者中断线程后线程终止并且没有成功中断线程
            //大家好好理解下这个逻辑
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                //中断线程
                wt.interrupt();
            try {
                //执行任务前执行的逻辑
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    //调用Runable接口的run方法执行任务
                    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;
                //完成的任务数量加1
                w.completedTasks++;
                //释放工作线程获得的锁
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //执行退出Worker线程的逻辑
        processWorkerExit(w, completedAbruptly);
    }
}

这里,我们拆解runWorker(Worker)方法。

(1)获取当前线程的句柄和工作线程中的任务,并将工作线程中的任务设置为空,执行unlock方法释放锁,将state状态设置为0,此时可以中断工作线程,代码如下所示。

Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
//释放锁,将state设置为0,允许中断任务的执行
w.unlock();

(2)在while循环中进行判断,如果任务不为空,或者从任务队列中获取的任务不为空,则执行while循环,否则,调用processWorkerExit(Worker, boolean)方法退出Worker工作线程。

while (task != null || (task = getTask()) != null)

(3)如果满足while的循环条件,首先获取工作线程内部的独占锁,并执行一系列的逻辑判断来检测是否需要中断当前线程的执行,代码如下所示。

//如果任务不为空,则获取Worker工作线程的独占锁
w.lock();
//如果线程已经停止,或者中断线程后线程终止并且没有成功中断线程
//大家好好理解下这个逻辑
if ((runStateAtLeast(ctl.get(), STOP) ||
     (Thread.interrupted() &&
      runStateAtLeast(ctl.get(), STOP))) &&
    !wt.isInterrupted())
    //中断线程
    wt.interrupt();

(4)调用执行任务前执行的逻辑,如下所示

//执行任务前执行的逻辑
beforeExecute(wt, task);

(5)调用Runable接口的run方法执行任务

//调用Runable接口的run方法执行任务
task.run();

(6)调用执行任务后执行的逻辑

//执行任务后执行的逻辑
afterExecute(task, thrown);

(7)将完成的任务设置为空,完成的任务数量加1并释放工作线程的锁。

//任务执行完成后,将其设置为空
task = null;
//完成的任务数量加1
w.completedTasks++;
//释放工作线程获得的锁
w.unlock();

(8)退出Worker线程的执行,如下所示

//执行退出Worker线程的逻辑
processWorkerExit(w, completedAbruptly);

从代码分析上可以看到,当从Worker线程中获取的任务为空时,会调用getTask()方法从任务队列中获取任务,接下来,我们看下getTask()方法的实现。

getTask()方法

我们先来看下getTask()方法的源代码,如下所示。

private Runnable getTask() {
    //轮询是否超时的标识
    boolean timedOut = false;
    //自旋for循环
    for (;;) {
        //获取ctl
        int c = ctl.get();
        //获取线程池的状态
        int rs = runStateOf(c);

        //检测任务队列是否在线程池停止或关闭的时候为空
        //也就是说任务队列是否在线程池未正常运行时为空
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            //减少Worker线程的数量
            decrementWorkerCount();
            return null;
        }
        //获取线程池中线程的数量
        int wc = workerCountOf(c);

        //检测当前线程池中的线程数量是否大于corePoolSize的值或者是否正在等待执行任务
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        //如果线程池中的线程数量大于corePoolSize
        //获取大于corePoolSize或者是否正在等待执行任务并且轮询超时
        //并且当前线程池中的线程数量大于1或者任务队列为空
        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()方法的逻辑比较简单,大家看源码就可以了,我这里就不重复描述了。

接下来,我们看下在正式调用Runnable的run()方法前后,执行的beforeExecute方法和afterExecute方法。

beforeExecute(Thread, Runnable)方法

beforeExecute(Thread, Runnable)方法的源代码如下所示。

protected void beforeExecute(Thread t, Runnable r) { }

可以看到,beforeExecute(Thread, Runnable)方法的方法体为空,我们可以创建ThreadPoolExecutor的子类来重写beforeExecute(Thread, Runnable)方法,使得线程池正式执行任务之前,执行我们自己定义的业务逻辑。

afterExecute(Runnable, Throwable)方法

afterExecute(Runnable, Throwable)方法的源代码如下所示。

protected void afterExecute(Runnable r, Throwable t) { }

可以看到,afterExecute(Runnable, Throwable)方法的方法体同样为空,我们可以创建ThreadPoolExecutor的子类来重写afterExecute(Runnable, Throwable)方法,使得线程池在执行任务之后执行我们自己定义的业务逻辑。

接下来,就是退出工作线程的processWorkerExit(Worker, boolean)方法。

processWorkerExit(Worker, boolean)方法

processWorkerExit(Worker, boolean)方法的逻辑主要是执行退出Worker线程,并且对一些资源进行清理,源代码如下所示。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    //执行过程中出现了异常,突然中断
    if (completedAbruptly)
        //将工作线程的数量减1
        decrementWorkerCount();
    //获取全局锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //累加完成的任务数量
        completedTaskCount += w.completedTasks;
        //将完成的任务从workers集合中移除
        workers.remove(w);
    } finally {
        //释放锁
        mainLock.unlock();
    }
    //尝试终止工作线程的执行
    tryTerminate();
    //获取ctl
    int c = ctl.get();
    //判断当前线程池的状态是否小于STOP(RUNNING或者SHUTDOWN)
    if (runStateLessThan(c, STOP)) {
        //如果没有突然中断完成
        if (!completedAbruptly) {
            //如果allowCoreThreadTimeOut为true,为min赋值为0,否则赋值为corePoolSize
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            //如果min为0并且工作队列不为空
            if (min == 0 && ! workQueue.isEmpty())
                //min的值设置为1
                min = 1;
            //如果线程池中的线程数量大于min的值
            if (workerCountOf(c) >= min)
                //返回,不再执行程序
                return; 
        }
        //调用addWorker方法
        addWorker(null, false);
    }
}

接下来,我们拆解processWorkerExit(Worker, boolean)方法。

(1)执行过程中出现了异常,突然中断执行,则将工作线程数量减1,如下所示。

//执行过程中出现了异常,突然中断
if (completedAbruptly)
    //将工作线程的数量减1
    decrementWorkerCount();

(2)获取锁累加完成的任务数量,并将完成的任务从workers集合中移除,并释放,如下所示。

//获取全局锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
    //累加完成的任务数量
    completedTaskCount += w.completedTasks;
    //将完成的任务从workers集合中移除
    workers.remove(w);
} finally {
    //释放锁
    mainLock.unlock();
}

(3)尝试终止工作线程的执行

//尝试终止工作线程的执行
tryTerminate();

(4)处判断当前线程池中的线程个数是否小于核心线程数,如果是,需要新增一个线程保证有足够的线程可以执行任务队列中的任务或者提交的任务。

//获取ctl
int c = ctl.get();
//判断当前线程池的状态是否小于STOP(RUNNING或者SHUTDOWN)
if (runStateLessThan(c, STOP)) {
    //如果没有突然中断完成
    if (!completedAbruptly) {
        //如果allowCoreThreadTimeOut为true,为min赋值为0,否则赋值为corePoolSize
        int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
        //如果min为0并且工作队列不为空
        if (min == 0 && ! workQueue.isEmpty())
            //min的值设置为1
            min = 1;
        //如果线程池中的线程数量大于min的值
        if (workerCountOf(c) >= min)
            //返回,不再执行程序
            return; 
    }
    //调用addWorker方法
    addWorker(null, false);
}

接下来,我们看下tryTerminate()方法。

tryTerminate()方法

tryTerminate()方法的源代码如下所示。

final void tryTerminate() {
    //自旋for循环
    for (;;) {
        //获取ctl
        int c = ctl.get();
        //如果线程池的状态为RUNNING
        //或者状态大于TIDYING
        //或者状态为SHUTDOWN并且任务队列为空
        //直接返回程序,不再执行后续逻辑
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        //如果当前线程池中的线程数量不等于0
        if (workerCountOf(c) != 0) { 
            //中断线程的执行
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        //获取线程池的全局锁
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //通过CAS将线程池的状态设置为TIDYING
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    //调用terminated()方法
                    terminated();
                } finally {
                    //将线程池状态设置为TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    //唤醒所有因为调用线程池的awaitTermination方法而被阻塞的线程
                    termination.signalAll();
                }
                return;
            }
        } finally {
            //释放锁
            mainLock.unlock();
        }
    }
}

(1)获取ctl,根据情况设置线程池状态或者中断线程的执行,并返回。

//获取ctl
int c = ctl.get();
//如果线程池的状态为RUNNING
//或者状态大于TIDYING
//或者状态为SHUTDOWN并且任务队列为空
//直接返回程序,不再执行后续逻辑
if (isRunning(c) ||
    runStateAtLeast(c, TIDYING) ||
    (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
    return;
//如果当前线程池中的线程数量不等于0
if (workerCountOf(c) != 0) { 
    //中断线程的执行
    interruptIdleWorkers(ONLY_ONE);
    return;
}

(2)获取全局锁,通过CAS设置线程池的状态,调用terminated()方法执行逻辑,最终将线程池的状态设置为TERMINATED,唤醒所有因为调用线程池的awaitTermination方法而被阻塞的线程,最终释放锁,如下所示。

//获取线程池的全局
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
    //通过CAS将线程池的状态设置为TIDYING
    if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
        try {
            //调用terminated()方法
            terminated();
        } finally {
            //将线程池状态设置为TERMINATED
            ctl.set(ctlOf(TERMINATED, 0));
            //唤醒所有因为调用线程池的awaitTermination方法而被阻塞的线程
            termination.signalAll();
        }
        return;
    }
} finally {
    //释放锁
    mainLock.unlock();
}

接下来,看下terminated()方法。

terminated()方法

terminated()方法的源代码如下所示。

protected void terminated() { }

可以看到,terminated()方法的方法体为空,我们可以创建ThreadPoolExecutor的子类来重写terminated()方法,值得Worker线程调用tryTerminate()方法时执行我们自己定义的terminated()方法的业务逻辑。

好了,今天就到这儿吧,我是冰河,我们下期见~~

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
JAVA多线程高并发学习笔记(三)——Callable、Future和FutureTask
为什么要是用Callable和Future Runnable的局限性 Executor采用Runnable作为基本的表达形式,虽然Runnable的run方法能够写入日志,写入文件,写入数据库等操作,但是它不能返回一个值,或者抛出一个受检查的异常,有些需要返回值的需求就不能满足了。
1233 0
Java多线程高并发学习笔记——阻塞队列
在探讨可重入锁之后,接下来学习阻塞队列,这篇文章也是断断续续的写了很久,因为最近开始学ssm框架,准备做一个自己的小网站,后续可能更新自己写网站的技术分享。 请尊重作者劳动成果,转载请标明原文链接: http://www.cnblogs.com/superfj/p/7757876.html 阻塞队列是什么? 首先了解队列,队列是数据先进先出的一种数据结构。
866 0
Java多线程高并发学习笔记(一)——Thread&Runnable
进程与线程 首先来看百度百科关于进程的介绍: 进程是一个具有独立功能的程序关于某个数据集合的一次运行活动。它可以申请和拥有系统资源,是一个动态的概念,是一个活动的实体。它不只是程序的代码,还包括当前的活动,通过程序计数器的值和处理寄存器的内容来表示。
945 0
JavaScript异步精讲,让你更加明白Js的执行流程!
JavaScript异步精讲,让你更加明白Js的执行流程! 问题点 什么是单线程,和异步有什么关系 什么是 event-loop jQuery的Deferred Promise 的基本使用和原理 async/await(和 Promise的区别、联系) 一、什么是单线程,和异步有什么关系 单线程.
1808 0
ArrayList为什么线程不安全 线程不安全体现在哪些方面 源码角度分析其具体原因
我们都知道ArrayList是线程不安全的,那么它不安全在哪里?又会出现什么并发问题呢?
218 0
+关注
冰河001
冰河,互联网高级技术专家,《深入理解分布式事务:原理与实战》,《海量数据处理与大数据技术实战》和《MySQL技术大全:开发、优化与运维实战》作者,mykit-data与mykit-transaction-message开源框架作者,公众号【冰河技术】作者。
29
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
文娱运维技术
立即下载
《SaaS模式云原生数据仓库应用场景实践》
立即下载
《看见新力量:二》电子书
立即下载