美文网首页
线程池底层原理

线程池底层原理

作者: 五月花号区块链联盟 | 来源:发表于2019-03-03 11:48 被阅读0次

    概述

    JAVA通过多线程的方式实现并发,为了方便线程池的管理,JAVA采用线程池的方式对线线程的整个生命周期进行管理。1.5后引入的Executor框架的最大优点是把任务的提交和执行解耦

    要执行任务的人只需把Task描述清楚,然后提交即可。这个Task是怎么被执行的,被谁执行的,什么时候执行的,提交的人就不用关心了。

    线程池同时可以避免创建大量线程的开销,提高响应速度。最近在阅读JVM相关的东西,一个对象的创建需要以下过程:

    1. 检查对应的类是否已经被加载、解析和初始化
    2. 类加载后,为新生对象分配内存
    3. 将分配到的内存空间初始为 0
    4. 对对象进行关键信息的设置,比如对象的hashcode等
    5. 然后执行 init 方法初始化对象

    如果每次都是如此的创建线程->执行任务->销毁线程,会造成很大的性能开销。复用已创建好的线程可以提高系统的性能,借助池化技术的思想,通过预先创建好多个线程,放在池中,这样可以在需要使用线程的时候直接获取,避免多次重复创建、销毁带来的开销。

    线程池的“池”

    ThreadPoolExecutor

    前面提到一个名词——池化技术,那么到底什么是池化技术呢?池化技术简单点来说,就是提前保存大量的资源,以备不时之需。在机器资源有限的情况下,使用池化技术可以大大的提高资源的利用率,提升性能等。

    在编程领域,比较典型的池化技术有:

    线程池、连接池、内存池、对象池等。

    在Java中创建线程池可以使用ThreadPoolExecutor,其继承关系如下图

    [图片上传失败...(image-cd5126-1551584999145)]

    其构造函数为:

    代码块

    Java

    public ThreadPoolExecutor(int corePoolSize,    //核心线程的数量
                              int maximumPoolSize,    //最大线程数量
                              long keepAliveTime,    //超出核心线程数量以外的线程空余存活时间
                              TimeUnit unit,    //存活时间的单位
                              BlockingQueue<Runnable> workQueue,    //保存待执行任务的队列
                              ThreadFactory threadFactory,    //创建新线程使用的工厂
                              RejectedExecutionHandler handler // 当任务无法执行时的处理器
                              ) {...}
    
    • corePoolSize:核心线程池数量

    在线程数少于核心数量时,有新任务进来就新建一个线程,即使有的线程没事干

    等超出核心数量后,就不会新建线程了,空闲的线程就得去任务队列里取任务执行了

    • maximumPoolSize:最大线程数量

    包括核心线程池数量 + 核心以外的数量

    如果任务队列满了,并且池中线程数小于最大线程数,会再创建新的线程执行任务

    • keepAliveTime:核心池以外的线程存活时间,即没有任务的外包的存活时间

    如果给线程池设置 allowCoreThreadTimeOut(true),则核心线程在空闲时头上也会响起死亡的倒计时

    如果任务是多而容易执行的,可以调大这个参数,那样线程就可以在存活的时间里有更大可能接受新任务

    • workQueue:保存待执行任务的阻塞队列

    不同的任务类型有不同的选择,下一小节介绍

    • threadFactory:每个线程创建的地方

    可以给线程起个好听的名字,设置个优先级啥的

    • handler:饱和策略,大家都很忙,咋办呢,有四种策略
      • AbortPolicy:直接抛出 RejectedExecutionException 异常,本策略也是默认的饱和策略
      • CallerRunsPolicy:只要线程池没关闭,就直接用调用者所在线程来运行任务
      • DiscardPolicy:悄悄把任务放生,不做了
      • DiscardOldestPolicy:把队列里待最久的那个任务扔了,然后再调用 execute() 尝试执行
      • 我们也可以实现自己的 RejectedExecutionHandler 接口自定义策略,比如如记录日志什么的

    如果把线程比作员工,那么线程池可以比作一个团队,核心池比作团队中正式员工数,核心池外的比作外包员工。

    线程池中任务的执行顺序

    通过Executors静态工厂也可以构建常用的线程池,在详细介绍之前,还需要先了解线程池中任务的执行顺序

        public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            /*
             * Proceed in 3 steps:
             *
             * 1. If fewer than corePoolSize threads are running, try to
             * start a new thread with the given command as its first
             * task.  The call to addWorker atomically checks runState and
             * workerCount, and so prevents false alarms that would add
             * threads when it shouldn't, by returning false.
             *
             * 2. If a task can be successfully queued, then we still need
             * to double-check whether we should have added a thread
             * (because existing ones died since last checking) or that
             * the pool shut down since entry into this method. So we
             * recheck state and if necessary roll back the enqueuing if
             * stopped, or start a new thread if there are none.
             *
             * 3. If we cannot queue task, then we try to add a new
             * thread.  If it fails, we know we are shut down or saturated
             * and so reject the task.
             */
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) {
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            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);
            }
            else if (!addWorker(command, false))
                reject(command);
        }
    

    从注释中可以看到处理逻辑,从判断条件中可以看到核心模块

    • 第一个红框:workerCountOf方法根据ctl的低29位,得到线程池的当前线程数,如果线程数小于corePoolSize,则执行addWorker方法创建新的线程执行任务;
    • 第二个红框:判断线程池是否在运行,如果在,任务队列是否允许插入,插入成功再次验证线程池是否运行,如果不在运行,移除插入的任务,然后抛出拒绝策略。如果在运行,没有线程了,就启用一个线程。
    • 第三个红框:如果添加非核心线程失败,就直接拒绝了。

    概略图:

    [图片上传失败...(image-65dd15-1551584999145)]

    详细流程图:

    [图片上传失败...(image-7d58a9-1551584999145)]

    Executors

    按照上面的总结,可以逐一分析Executors工厂类提供的现成的线程池:

    [图片上传失败...(image-64c0c6-1551584999145)]

    1.newFixedThreadPool

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    

    不招外包,有固定数量核心成员的正常互联网团队。

    可以看到,FixedThreadPool 的核心线程数和最大线程数都是指定值,也就是说当线程池中的线程数超过核心线程数后,任务都会被放到阻塞队列中。

    此外 keepAliveTime 为 0,也就是多余的空余线程会被立即终止(由于这里没有多余线程,这个参数也没什么意义了)。

    而这里选用的阻塞队列是 LinkedBlockingQueue,使用的是默认容量 Integer.MAX_VALUE,相当于没有上限。

    因此这个线程池执行任务的流程如下:

    线程数少于核心线程数,也就是设置的线程数时,新建线程执行任务

    线程数等于核心线程数后,将任务加入阻塞队列

    由于队列容量非常大,可以一直加加加

    执行完任务的线程反复去队列中取任务执行

    FixedThreadPool 用于负载比较重的服务器,为了资源的合理利用,需要限制当前线程数量。

    2.newSingleThreadExecutor

    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    

    不招外包,只有一个核心成员的创业团队。

    从参数可以看出来,SingleThreadExecutor 相当于特殊的 FixedThreadPool,它的执行流程如下:

    线程池中没有线程时,新建一个线程执行任务

    有一个线程以后,将任务加入阻塞队列,不停加加加

    唯一的这一个线程不停地去队列里取任务执行

    听起来很可怜的样子 - -。

    SingleThreadExecutor 用于串行执行任务的场景,每个任务必须按顺序执行,不需要并发执行。

    3.newCachedThreadPool

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    

    全部外包,没活最多待 60 秒的外包团队。

    可以看到,CachedThreadPool 没有核心线程,非核心线程数无上限,也就是全部使用外包,但是每个外包空闲的时间只有 60 秒,超过后就会被回收。

    CachedThreadPool 使用的队列是 SynchronousQueue,这个队列的作用就是传递任务,并不会保存。

    因此当提交任务的速度大于处理任务的速度时,每次提交一个任务,就会创建一个线程。极端情况下会创建过多的线程,耗尽 CPU 和内存资源。

    它的执行流程如下:

    没有核心线程,直接向 SynchronousQueue 中提交任务

    如果有空闲线程,就去取出任务执行;如果没有空闲线程,就新建一个

    执行完任务的线程有 60 秒生存时间,如果在这个时间内可以接到新任务,就可以继续活下去,否则就拜拜

    由于空闲 60 秒的线程会被终止,长时间保持空闲的 CachedThreadPool 不会占用任何资源。

    CachedThreadPool 用于并发执行大量短期的小任务,或者是负载较轻的服务器。

    4.newScheduledThreadPool

    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
    
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE,
              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
              new DelayedWorkQueue());
    }
    private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;
    

    定期维护的 2B 业务团队,核心与外包成员都有。

    ScheduledThreadPoolExecutor 继承自 ThreadPoolExecutor, 最多线程数为 Integer.MAX_VALUE ,使用 DelayedWorkQueue 作为任务队列。

    ScheduledThreadPoolExecutor 添加任务和执行任务的机制与ThreadPoolExecutor 有所不同。

    ScheduledThreadPoolExecutor 添加任务提供了另外两个方法:

    scheduleAtFixedRate() :按某种速率周期执行

    scheduleWithFixedDelay():在某个延迟后执行

    它俩的代码如下:

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        if (command == null || unit == null)
          throw new NullPointerException();
        if (period <= 0L)
          throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
          new ScheduledFutureTask<Void>(command,
                                        null,
                                        triggerTime(initialDelay, unit),
                                        unit.toNanos(period),
                                        sequencer.getAndIncrement());
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }
    
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)
          throw new NullPointerException();
        if (delay <= 0L)
          throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
          new ScheduledFutureTask<Void>(command,
                                        null,
                                        triggerTime(initialDelay, unit),
                                        -unit.toNanos(delay),
                                        sequencer.getAndIncrement());
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }
    

    可以看到,这两种方法都是创建了一个 ScheduledFutureTask 对象,调用 decorateTask() 方法转成 RunnableScheduledFuture 对象,然后添加到队列中。

    看下 ScheduledFutureTask 的主要属性:

    private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V> {
        //添加到队列中的顺序
        private final long sequenceNumber;
        //何时执行这个任务
        private volatile long time;
        //执行的间隔周期
        private final long period;
        //实际被添加到队列中的 task
        RunnableScheduledFuture<V> outerTask = this;
        //在 delay queue 中的索引,便于取消时快速查找
        int heapIndex;
        //...
    }
    

    DelayQueue 中封装了一个优先级队列,这个队列会对队列中的 ScheduledFutureTask 进行排序,两个任务的执行 time 不同时,time 小的先执行;否则比较添加到队列中的顺序 sequenceNumber ,先提交的先执行。

    ScheduledThreadPoolExecutor 的执行流程如下:

    调用上面两个方法添加一个任务

    线程池中的线程从 DelayQueue 中取任务

    然后执行任务

    具体执行任务的步骤也比较复杂:

    线程从 DelayQueue 中获取 time 大于等于当前时间的 ScheduledFutureTask

    DelayQueue.take()

    执行完后修改这个 task 的 time 为下次被执行的时间

    然后再把这个 task 放回队列中

    DelayQueue.add()

    ScheduledThreadPoolExecutor 用于需要多个后台线程执行周期任务,同时需要限制线程数量的场景。

    ”不允许使用“Executors

    阿里巴巴Java开发手册中明确指出,『不允许』使用Executors创建线程池。
    [图片上传失败...(image-6b63e2-1551584999145)]
    通过上面的例子,我们知道了Executors创建的线程池存在OOM的风险,那么到底是什么原因导致的呢?我们需要深入Executors的源码来分析一下。

    其实,在上面的报错信息中,我们是可以看出蛛丝马迹的,在以上的代码中其实已经说了,真正的导致OOM的其实是LinkedBlockingQueue.offer方法。

    Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded
        at java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:416)
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1371)
        at com.hollis.ExecutorsDemo.main(ExecutorsDemo.java:16)
    
    

    如果对Java中的阻塞队列有所了解的话,看到这里或许就能够明白原因了。

    Java中的BlockingQueue主要有两种实现,分别是ArrayBlockingQueue 和 LinkedBlockingQueue。

    ArrayBlockingQueue是一个用数组实现的有界阻塞队列,必须设置容量。

    LinkedBlockingQueue是一个用链表实现的有界阻塞队列,容量可以选择进行设置,不设置的话,将是一个无边界的阻塞队列,最大长度为Integer.MAX_VALUE。

    这里的问题就出在:不设置的话,将是一个无边界的阻塞队列,最大长度为Integer.MAX_VALUE。也就是说,如果我们不设置LinkedBlockingQueue的容量的话,其默认容量将会是Integer.MAX_VALUE。

    而newFixedThreadPool中创建LinkedBlockingQueue时,并未指定容量。此时,LinkedBlockingQueue就是一个无边界队列,对于一个无边界队列来说,是可以不断的向队列中加入任务的,这种情况下就有可能因为任务过多而导致内存溢出问题。

    上面提到的问题主要体现在newFixedThreadPool和newSingleThreadExecutor两个工厂方法上,并不是说newCachedThreadPool和newScheduledThreadPool这两个方法就安全了,这两种方式创建的最大线程数可能是Integer.MAX_VALUE,而创建这么多线程,必然就有可能导致OOM。

    说回ThreadPoolService

    addWorker

    从方法execute的实现可以看出:addWorker主要负责创建新的线程并执行任务,代码如下(这里代码有点长,没关系,也是分块的,总共有5个关键的代码块):

    [图片上传失败...(image-fb1f5d-1551584999145)]

    • 第一个红框:做是否能够添加工作线程条件过滤:
      • 判断线程池的状态,如果线程池的状态值大于或等SHUTDOWN,则不处理提交的任务,直接返回;
    • 第二个红框:做自旋,更新创建线程数量:
      • 通过参数core判断当前需要创建的线程是否为核心线程,如果core为true,且当前线程数小于corePoolSize,则跳出循环,开始创建新的线程。retry 是什么?这个是java中的goto语法。只能运用在break和continue后面。

    接着看后面的代码:

    [图片上传失败...(image-e0cba0-1551584999145)]

    • 第一个红框:获取线程池主锁。
      • 线程池的工作线程通过Woker类实现,通过ReentrantLock锁保证线程安全。
    • 第二个红框:添加线程到workers中(线程池中)。
    • 第三个红框:启动新建的线程。

    接下来,我们看看workers是什么。

    [图片上传失败...(image-da7e1b-1551584999145)]

    一个hashSet。所以,线程池底层的存储结构其实就是一个HashSet

    worker线程处理队列任务

    [图片上传失败...(image-116e1a-1551584999145)]

    • 第一个红框:是否是第一次执行任务,或者从队列中可以获取到任务。
    • 第二个红框:获取到任务后,执行任务开始前操作钩子。
    • 第三个红框:执行任务。
    • 第四个红框:执行任务后钩子。

    这两个钩子(beforeExecute,afterExecute)允许我们自己继承线程池,做任务执行前后处理。

    总结

    到这里,源代码分析到此为止。接下来做一下简单的总结。

    所谓线程池本质是一个hashSet。多余的任务会放在阻塞队列中。

    只有当阻塞队列满了后,才会触发非核心线程的创建。所以非核心线程只是临时过来打杂的。直到空闲了,然后自己关闭了。

    线程池提供了两个钩子(beforeExecute,afterExecute)给我们,我们继承线程池,在执行任务前后做一些事情。

    线程池原理关键技术:锁(lock,cas)、阻塞队列、hashSet(资源池)

    [图片上传失败...(image-d32ae6-1551584999145)]

    参考文档

    Java中线程池,你真的会用吗?

    深入源码分析Java线程池的实现原理

    [线程池的使用与执行流程

    相关文章

      网友评论

          本文标题:线程池底层原理

          本文链接:https://www.haomeiwen.com/subject/nvspuqtx.html