线程池复用原理

更多精彩请关注公众号xhJaver,京东工程师和你一起成长

一、线程池状态

首先我们要明确线程池的几种状态

1. RUNNING

  • 这个状态表明线程池处于正常状态,可以处理任务,可以接受任务

2. SHUTDOWN

  • 这个状态表明线程池处于正常关闭状态,不再接受任务,但是可以处理线程池中剩余的任务

3. STOP

  • 这个状态表明线程池处于停止状态,不仅不会再接收新任务,并且还会打断正在执行的任务

4. TIDYING

  • 这个状态表明线程池已经没有了任务,所有的任务都被停掉了

5. TERMINATED

  • 线程池彻底终止状态

他们的状态转换图如下

线程池状态.png

<figcaption style="margin-top: 5px; text-align: center; color: #888; font-size: 14px;">线程池状态</figcaption>

好了,知道了线程池的几种状态和他们是如何转换的关系之后,我们来看一下 当我们提交一个任务时,线程池到底发生了什么?!

我们平常使用线程池是这样使用的

 for (int i=0;i<10;i++){
            //创建10个任务
            Task task = new Task("task" + i);
            //让我们自定义的线程池去跑这些任务
            threadPoolExecutor.execute(task);
        }

我们来看一下 execute里面究竟有什么奇怪的东西?

二、execute源码

  public void execute(Runnable command) {
        //1.先判断提交的任务是不是空的
        if (command == null)
            throw new NullPointerException();
        //2.获得线程池状态
        int c = ctl.get();
        //3.判断线程池数量是否小于核心线程数
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //4.线程池数量大于等于核心线程数并且线程池处于Running状态,这时添加任务至阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 5.走到这里说明添加阻塞队列失败,
        // 创建非核心线程也失败的话,执行拒绝策略
        else if (!addWorker(command, false))
            reject(command);
    }

然后我们来看3那里,

 //3.判断线程池数量是否小于核心线程数
        if (workerCountOf(c) < corePoolSize) {
            //如果小于核心线程数
            //3.1添加worker
            if (addWorker(command, true))
                //添加成功,返回
                return;
            //3.2添加失败,获取线程池状态
            c = ctl.get();
        }

我用头发想想都知道,线程复用的秘密肯定藏在了addworker里,哦对我没有头发 我们再来看一看他里面有什么鬼

三、addworker源码

 private boolean addWorker(Runnable firstTask, boolean core) {
      //标志位,一会儿会跳过来
        retry:
        for (;;) {
            //判断线程池状态
            int c = ctl.get();
            int rs = runStateOf(c);
            //如果状态非法则返回false
            if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
                return false;

            for (;;) {
                //判断线程池线程总数量
                int wc = workerCountOf(c);
                //如果数量不符合要求则返回false
                if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //线程池数量加1
                if (compareAndIncrementWorkerCount(c))
                    //跳到开始 retry处且往下执行,不在进入大循环
                    break retry;
                //数量增加失败的话判断当前线程池状态若和刚才状态不一致则继续执行大循环
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            //将提交的任务封装进worker
            w = new Worker(firstTask);
            //得到worker中的线程
            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)) {
                        if (t.isAlive())
                            throw new IllegalThreadStateException();
                        //将worker添加至workers中 (这是个set集合,真正的线程池) 
                        workers.add(w);
                        //判断线程数量
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        //添加worker成功
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    //执行worker中的线程
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

其中很重要的一段代码是

 //将提交的任务封装进worker
            w = new Worker(firstTask);
            //得到worker中的线程
            final Thread t = w.thread;
            ....
            .....
              //执行worker中的线程
                    t.start();

主要我们看这其中的worker是什么东西 (截取了worker中一部分源码)

四、Worker源码

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {

        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;

        Worker(Runnable firstTask) {
            //线程池状态设为running
            setState(-1); // inhibit interrupts until runWorker
            //用户提交的任务
            this.firstTask = firstTask;
//通过创建一个线程,传入的this是woker自身  worker继承了Runnable 那么这个线程在t.start就是调用重写的run()方法了
            this.thread = getThreadFactory().newThread(this);
        }

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

我们注意到刚才的t.start(); 就是执行woker中的run方法,run方法又执行了runworker() 方法 我们再来看下 runworker() 方法

五、runworker源码

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循环,判断如果提交的任务不为空或者队列里有任务的话
            while (task != null || (task = getTask()) != null) {
                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 {
                        //任务自己的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;
                    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);

            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            //得到线程池线程数量 
            int wc = workerCountOf(c);
            // 是否设置超时时间  allowCoreThreadTimeOut默认是false   
            //判断线程池数量是否大于核心线程数,如果大于的话  timed为true
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

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

            try {
                //这里 timed为ture的时候,采用带超时时间的获取元素的方法, 否则采取一直阻塞的方法
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                //获取到任务就返回
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

由此可见 getTask里面是有超时标志的timed的,我们在第一篇平常说的线程池原理里面讲到,若非核心线程空闲keepAliveTime分钟则销毁,就是在这里,keepAliveTime时间内未获取到任务,即为线程空闲状态,就退出了runWorker中的while循环,进行销毁线程的操作。

核心线程一定不会销毁吗? 我们注意到,这里面有一个allowCoreThreadTimeOut变量,如果他要是为true的话,那么核心线程也是可以销毁的

threadPoolExecutor.allowCoreThreadTimeOut(true);

真的有核心线程与非核心线程之分吗? 其实是没有区别的,他们都是一样的线程,线程池源码中并没有核心线程这个标记,只是有一个核心线程数量,在这个数量之前创建先线程和在这个数量之后创建线程,默认在这个数量之后创建的线程会在keepAliveTime空闲时间内销毁,我们为了方便记忆,而将其称为非核心线程

七、总结

大体流程如下图所示


线程池源码流程图.PNG

我们向线程池提交任务后,线程池会将我们的任务封装成一个worker,这个worker里面有要执行的线程t和要执行的任务,这个线程t的主要任务就是t.start,运行runworker方法,在runworker方法中,会一直while循环获取提交的任务若没有提交的任务则会看队列里有没有任务,获取队列任务时就会判断超时标志是否为true,如果为true的话,则在超时时间内未获取到任务则返回null,然后销毁当前线程,否则一直等待到获取到任务为止,不销毁线程,这样就做到了线程复用。

更多精彩请关注公众号xhJaver,京东工程师和你一起成长

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 213,186评论 6 492
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,858评论 3 387
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 158,620评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,888评论 1 285
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 66,009评论 6 385
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 50,149评论 1 291
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,204评论 3 412
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,956评论 0 268
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,385评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,698评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,863评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,544评论 4 335
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,185评论 3 317
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,899评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,141评论 1 267
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,684评论 2 362
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,750评论 2 351