文档章节

ThreadPoolExecutor源码解析(二)

终日而思一
 终日而思一
发布于 11/19 17:53
字数 2069
阅读 4
收藏 1

1.ThreadPoolExcuter运行实例

  首先我们先看如何新建一个ThreadPoolExecutor去运行线程。然后深入到源码中去看ThreadPoolExecutor里面使如何运作的。

public class Test {
    public static void main(String[] args){
        /**
         * 新建一个线程池
         * corePoolSize:2
         * maximumPoolSize:10
         * keepAliveTime:20
         * unit:TimeUnit.SECONDS(秒)
         * workQueue:new ArrayBlockingQueue(10)
         * threadFactory:默认
         * RejectedExecutionHandler默认
         */
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2,10,20, TimeUnit.SECONDS,new ArrayBlockingQueue(10));
        //用execute添加一个线程
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

2.ThreadPoolExecute.execute方法

  可以发现,其实使用线程池就是使用这个方法,然后我们看这个方法具体的代码。


    /**
     * 在后面执行给定任务。任务在一个新的线程中或一个存在的worker的线程池中执行。
     * 如果一个线程不能提交到excution,可能是因为这个excutor已经shundown或者因为其容量已经是最大,
     * 此时任务将会被RejectedExecutionHandler处理
     *
     */
    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         * 有以下3个步骤
         *
         * 1.如果少于corePoolSize的线程在运行,那么试着启动一个新线程,其中用给定指令作为first task。
         * 这会调用addWorker去原子性得检查runState和workerCoune,因此可以防止错误报警,在错误报警不应该时通过返回false来添加线程
         * 2.如果任务被成功排队,我们任然应该第二次检查是否添加一个新线程(因为可能存在在最后一次检查后挂掉的情况)
         * 或者在进入这个方法期间线程池shutdown。所以我们再次检查状态,如果已关闭和有必要则退出队列,或者如果没有的话就开始一个新的线程。
         * 3.如果我们无法将task入队,那么我们试图添加新线程。如果失败,那么知道我们shutdown或者是饱和的并拒绝task。
         */
        int c = ctl.get();
        //判断是否小于corePoolSize
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //如果pool在运行并且能提交到队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //这里进行再次检查,如果线程池没在运行并且成功删除task后,使用拒绝策略拒绝该task
            if (! isRunning(recheck) && remove(command))
                reject(command);
       //如果已经将task添加到队列中,而此时没有worker的话,那么新建一个worker。稍后这个空闲的worker就会自动去队列里面取任务来执行
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //如果无法提交那么按照拒绝策略拒绝task
        else if (!addWorker(command, false))
            reject(command);
    }

  线程池的基础和Worker基本介绍在前一节已经有说过,可以点这里查看。可以看到这个方法的主要流程,其实都在注释里面说明了。可以发现里面主要调用了一个方法,addWorker()。 那么这个addWorker()又是什么东西呢。其实看方法名就很清楚了,就是新建一个Worker来执行你添加进来的task。

3.ThreadPoolExecute.addWorker()方法

/**
     * 检查当前的线程池状态和容量,是否可以让一个新的worker加入。如果可以,worker计数将会被调整,并且
     * 如果可能,一个新的woker将会被创建和开始,将它当作第一个任务来运行。当线程池是stopped或shutdown状态时,
     * 将返回false。当线程工厂创建失败而返回null或者抛出exception(比如典型的OOM)时,它也会返回fails。
     * firstTask:新线程应该第一个运行的任务。当线程数少于corePoolSize时或是队列满时,workers使用一个初始化的
     * first task来创建,用来进行分流。初始化空闲线程通常使用prestartCoreThread。
     * core:为true,如果使用有界的corePoolSize,否则时maxPoolSize
     * @return true if successful
     * 添加Worker
     */
    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;
                //用到了原子CAS方法比较,使用CAS增加worker计数器成功,才能进入下一步
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                //重新获取ctl
                c = ctl.get();  // Re-read ctl
                //这里表示执行到这里的时候线程池的运行状态改变,需要重新跳到retry处执行
                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 {
            //使用firstTask初始化Worker,first可能为null,那么则表示该worker为空闲
            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();
                        //largestPoolSize为跟踪的目前最大线程数,因为之前已经做过判断,所以不会越界问题
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //workerAdded是在上面最后才设置的,确保这个变量能准确表示是否添加worker成功
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //再次检查
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

  addWorker本事只是为线程池添加一个Worker,其本身所做的事情其实很简单,但难就难在要确保安全有效得添加一个Worker。为此addWorker()方法做了很多额外的工作。比如判断线程池的运行状态,当前Worker数量是否已经饱和等等。可以发现在这个方法,或者说整个ThreadPoolExecutor中,很多时候都是使用双重检查的方式来对线程池状态进行检查。其实这都是为了效率,最简单不过直接使用Synchronized或ReentranLock进行同步,但这样效率会低很多,所以在这里,只有在万不得已的情况下,才会使用悲观的ReentranLock。

  addWorker的最后直接调用了t.start,这里的t其实就是Worker它自己。接下来再看Worker是如何运行的。

4.ThreadPoolExecute.runWorker()方法

/**
     * 主要的Worker运行的循环。重复得获取从任务队列中取出task并执行它。
    */

    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        //取出firstTask,再将worker中的值-设置为null
        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
                //如果当前线程是stop,那么将确认其为interrupted
                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);
        }
    }

  从源码中可以看出,一个Worker的工作其实就是不断使用getTask()方法从队列中获取新的任务来执行。值得一提的是,初始化参数里面的时间戳参数就是在这个方法里面运用的。在循环体中每次都使用锁以保证当前worker在运行task过程中不会被中断。同时运行时还会去调用两个内置的钩子:beforeExecute()和afterExecute(),这两个方法默认实现时空的。

  同时在运行的循环中每次都关注着ThreadPoolExecutor的运行状态,当线程池处于中断状态时,循环Worker的当前线程也会中断。

总结:说到这里就差不多把线程池运行task的流程说完了,当然其中忽略了很多的细节。但总而言之,ThreadPoolExecutor其实就是对worker进行管理,然后使用这些worker来执行用户提交的task。对用户提交的task的数量也进行一定的控制管理,比如超过一定数量时放入一个任务队列中等等。然后对线程池规定一些状态量,根据这些状态量对线程池进行控制。

© 著作权归作者所有

共有 人打赏支持
终日而思一
粉丝 1
博文 20
码字总数 33151
作品 0
广州
私信 提问
刨根问底(二):ThreadPoolExecutor

一、什么是ThreadPoolExecutor ThreadPoolExecutor是Java 1.5开始引入的,作为线程存放的集合池子——线程池,主要是为了解决: 重用线程资源,降低线程创建和销毁的开销; 集中维护和管理多...

叫我宫城大人
08/08
0
0
ThreadPoolExecutor JAVA1.8源码解析

1. 构造函数 corePoolSize:池里维持的最小线程数,即使它们是空闲线程,也不会进行销毁 maximumPoolSize:最大线程数 keepAliveTime:当池里的线程数量超过了corePoolSize时,如果额外线程在...

pomer_huang的博客
2017/12/22
0
0
Java多线程之线程池(ThreadPoolExecutor)实现原理分析(一)

在上一篇文章Java中实现多线程的3种方法介绍和比较中,我们讲解了Java中实现多线程的3种方法。使用多线程,就必须要考虑使用线程池,今天我们来聊聊线程池的那些事。 注:源码都是基于JDK1....

小怪聊职场
05/14
0
0
Java核心(二)深入理解线程池ThreadPool

本文你将获得以下信息: 线程池源码解读 线程池执行流程分析 带返回值的线程池实现 延迟线程池实现 为了方便读者理解,本文会由浅入深,先从线程池的使用开始再延伸到源码解读和源码分析等高...

王磊的博客
11/19
0
0
OkHttp 知识梳理(2) - OkHttp 源码解析之异步请求 & 线程调度

OkHttp 知识梳理系列文章 OkHttp 知识梳理(1) - OkHttp 源码解析之入门 OkHttp 知识梳理(2) - OkHttp 源码解析之异步请求 & 线程调度 OkHttp 知识梳理(3) - OkHttp 之缓存基础 一、前言 在 ...

泽毛
2017/11/22
0
0

没有更多内容

加载失败,请刷新页面

加载更多

【58沈剑 架构师之路】1分钟了解MyISAM与InnoDB的索引差异

《数据库索引,到底是什么做的?》介绍了B+树,它是一种非常适合用来做数据库索引的数据结构: (1)很适合磁盘存储,能够充分利用局部性原理,磁盘预读; (2)很低的树高度,能够存储大量数据;...

张锦飞
8分钟前
1
0
代码优化----使用builder模式构造对象

看《effective java》的时候,创建对象章节提到使用builder模式来创建对象。觉得非常好用,做一个记录。以后应该就会这么写啦~~~~~~ 对于一个有很多属性的类,在为属性赋值时,通常会用到两种...

wuyiyi
9分钟前
2
0
一文带你看懂cookie,面试前端不用愁

本文由云+社区发表 在前端面试中,有一个必问的问题:请你谈谈cookie和localStorage有什么区别啊? localStorage是H5中的一种浏览器本地存储方式,而实际上,cookie本身并不是用来做服务器存...

腾讯云加社区
10分钟前
0
0
随行付微服务测试之接口测试和契约测试

背景 日常开发过程中,项目的接口通常由服务提供方约定和提供,微服务模式下接口被多个消费者调用更是常态,那么提供方接口的变更如何快速、高效、无遗漏的通知给消费者呢?另外,当一个ser...

马力-随行付
11分钟前
1
0
为什么Python是2019最值得学的编程语言?

对于那些从来没有学习编程小伙伴,Python 是最好的选择之一, Python 是一种清晰的语言,用缩进来表示程序的嵌套关系可谓是一种创举,把过去软性的编程风格升级为硬性的语法规定。再不需要在...

糖宝lsh
11分钟前
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部