专栏首页一猿小讲Java线程池深度揭秘

Java线程池深度揭秘

作为 Java 程序员,无论是技术面试、项目研发或者是学习框架源码,不彻底掌握 Java 多线程的知识,做不到心中有数,干啥都没底气,尤其是技术深究时往往略显发憷。

坐稳扶好,通过今天的分享,能让你轻松 get 如下几点。

1. Executor 框架家族简介; 2. 源码解读:线程池状态以及状态流转; 3. 源码解读:部分成员变量及方法; 4. 源码解读:任务提交submit方法背后; 5. 源码揭秘之后的反思; 6. 寄语。

1

Excutor 家族简介

一图胜千言,脑中有图心不慌。

(一)Executor 接口。

public interface Executor {
    void execute(Runnable command);
}

Executor 是一个接口(主要用于定义规范),定义了 execute 方法,用于接收 Runnable 对象。

(二)ExecutorService 接口。

public interface ExecutorService extends Executor {
    // ... ...
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);
    // ... ...
}

ExecutorService 也是一个接口,继承了 Executor 接口,增加了更多方法,相当于扩展了 Executor 接口的功能,例如定义了 submit() 系列方法,支持任务执行后得到返回结果。

(三)AbstractExecutorService 抽象类。

public abstract class AbstractExecutorService implements ExecutorService {
    // ... ...
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }

    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }

    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }

    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
    // ... ...
}

AbstractExecutorService 是一个抽象类,实现了 ExecutorService 接口中的部分方法,例如提供了任务提交的 submit 方法的默认实现,而 submit 方法最终会调用 execute 方法。

不过 AbstractExecutorService 并没有实现 execute 方法,相当于为子类留了个口子,让子类去灵活扩展(钩子函数)。

(四)ScheduledExecutorService 接口。

public interface ScheduledExecutorService extends ExecutorService {
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);

    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);

    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
}

ScheduledExecutorService 接口继承了 ExecutorService,增加定时调度的方法,使其成为一个可定时调度任务的接口,相当于扩展了 ExecutorService 的功能。

(五)ScheduledThreadPoolExecutor 类。

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {
    // ... ...
}

ScheduledThreadPoolExecutor 类继承自 ThreadPoolExecutor 类,并且实现了 ScheduledExecutorService 接口,变成一个可定时调度任务的线程池。

(六)ThreadPoolExecutor 类。

public class ThreadPoolExecutor extends AbstractExecutorService {
    // ... ...
}

ThreadPoolExecutor 继承 AbstractExecutorService 抽象类,并实现了 execute 等一系列方法。

(七)Executors 类。

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

研发人员可以通过 Executors 工厂类来创建线程池并返回一个ExecutorService 对象,而内部几乎全是对 ThreadPoolExecutor 的封装。

通过 Executor 的家族简单认识,应该能感觉到 ThreadPoolExecutor 类的重要性,所以接下来要重点对 ThreadPoolExecutor 类的源码进行剖析。

2

源码解读:线程池状态以及状态流转

上面注释截图来源于 ThreadPoolExecutor 的源码,别懵圈,仔细看差不多都能懂,能够看出线程池的五种状态以及对应的状态流转。

不知道你能看懂多少,看不懂也没关系,接下来把上面的注释用图呈现给大家。通过源码中的注释,能够勾勒出如下线程池的状态流转图(好的注释是多么的重要啊,感叹号!)。

3

源码解读:部分成员变量及方法

/**
 * ctl 是一个 AtomicInteger 类型的原子对象。
 * 其实设计很有意思:ctl 共包括 32 位(高 3 位表示"线程池状态",低 29 位表示"线程池中的线程数量")。
 * 个人感觉:线程池状态与线程数量合二为一,用一个变量来表示,来减少锁竞争,提高并发效率。
 */
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
/** 表示线程池线程数的位数:32 - 3 = 29 位 */
private static final int COUNT_BITS = Integer.SIZE - 3;
/** 表示最大线程容量(000,11111111111111111111111111111)*/
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits(运行状态保存在 int 值的高 3 位)
/** 111,00000000000000000000000000000 */
private static final int RUNNING    = -1 << COUNT_BITS;
/** 000,00000000000000000000000000000 */
private static final int SHUTDOWN   =  0 << COUNT_BITS;
/** 001,00000000000000000000000000000 */
private static final int STOP       =  1 << COUNT_BITS;
/** 010,00000000000000000000000000000 */
private static final int TIDYING    =  2 << COUNT_BITS;
/** 011,00000000000000000000000000000 */
private static final int TERMINATED =  3 << COUNT_BITS;

// Packing and unpacking ctl
/** 获取线程池的运行状态 */
private static int runStateOf(int c)     { return c & ~CAPACITY; }
/** 线程池内有效线程的数量 (workerCount) */
private static int workerCountOf(int c)  { return c & CAPACITY; }
/** 线程池的状态和线程的数量组装,成为 ctl */
private static int ctlOf(int rs, int wc) { return rs | wc; }

仔细去看上面的代码,注释已经很清晰啦。重点关注 ctl 变量,这个变量将线程池自身状态和线程数量,融合在这一个变量中,其中高 3 位表示线程池状态,低 29 位表示线程池中的线程数量,这样在多线程环境下更易保证线程池自身状态和线程数量的统一,不得不佩服源代码作者 Doug Lea,可谓是设计甚妙!

4

源码解读:任务提交 submit 方法背后

疑问?当调用 submit() 方法,把一个任务提交给线程池去处理的时候,线程池的处理过程是什么样的呢?

通过开篇对 Executor 的家族简介,能够看到 submit() 方法最终会调用 ThreadPoolExecutor 的 execute 方法,走进源码好好看看 execute 方法都做了啥?

重点关注源码中的注释(红框圈住部分),若看懂注释,提交任务时线程池对应的处理,也就懂了一半啦(感触:好的编码规范真的好重要,业务开发时,核心代码一定要有注释)。

若依然很懵逼,一图胜千言,那就继续上图吧。

了解上图的整体流程,再去看看源码就彻悟啦。

public void execute(Runnable command) {
    //【Step 0. 如果任务为空则抛出 NPE 异常】
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    //【Step 1. 判断核心线程是否已满】
    // 1.1. 判断当前线程数是否已经达到核心线程数的限制
    if (workerCountOf(c) < corePoolSize) {
        // 1.2. 如果未达到核心线程数的限制,则会直接添加一个核心线程,并指定首次执行的任务,进行任务处理
        if (addWorker(command, true))
            return;
        // 1.3. 如果添加失败,则刷新线程池的状态和线程的数量对应的变量 ctl
        c = ctl.get();
    }
    //【Step 2. 判断阻塞队列是否已满】
    // 2.1. 检查线程池是否是运行状态,然后将任务添加到等待队列
    if (isRunning(c) && workQueue.offer(command)) {
        // 2.2. 任务成功添加到等待队列,再次刷新 ctl
        int recheck = ctl.get();
        // 2.3. 添加任务到等待队列成功后,如果线程池不是运行状态,则将刚添加的任务从队列移除并执行拒绝策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 2.4. 判断当前线程数量,如果线程数量为 0,则添加一个非核心线程,并且不指定首次执行任务
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    //【Step 3. 判断最大线程数量是否已经达到】
    // 3.1. 添加非核心线程,指定首次执行任务,如果添加失败,执行异常策略
    else if (!addWorker(command, false))
        reject(command);
}

结合注释去读代码,应该都能搞懂。很显然 execute 方法中,多处都调用了 addWorker 方法,接下来简单剖析一下 addWorker 方法。

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        // ... ...
        for (;;) {
            // ... ...
            // 通过 CAS 自旋,增加线程数 +1,增加成功跳出双层循环,继续往下执行
            if (compareAndIncrementWorkerCount(c))
                break retry;
            // ... ...
        }
    }
    // 到这儿,说明已经成功的将线程数 +1 了,但是真正的线程还没有被添加
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 添加线程,Worker 是继承了 AQS,实现了 Runnable 接口的包装类
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // ... ...
                // 添加新增的 Worker
                workers.add(w);
                int s = workers.size();
                if (s > largestPoolSize)
                    largestPoolSize = s;
                workerAdded = true;
                // ... ...
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                // 启动 Worker
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

为了简明扼要,方法酌情进行了删减。addWorker 方法主要是通过双重 for 循环进行线程数 +1,然后创建 Worker,并进行添加到 HashSet<Worker> workers 列表中,然后调用 t.start() 启动 Worker。

那么接下来再一起看看 Worker 里面都做了啥?

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable {
    // ... ...
    final Thread thread;
    Runnable firstTask;

    /**
     * 通过指定的 firstTask 任务创建 Worker 对象
     */
    Worker(Runnable firstTask) {
        setState(-1);
        this.firstTask = firstTask;
        // 通过当前 Worker 对象创建对应的线程对象 t,
        // 所以调用 t.start() 时最终会调用 Worker 的 run 方法
        this.thread = getThreadFactory().newThread(this);
    }

    public void run() {
        // run 方法最终会调用 ThreadPoolExecutor 的 runWorker 方法
        runWorker(this);
    }
    // ... ...
}

通过 Worker 的构造函数能够了解到,会通过创建的 Worker 对象去构建线程对象,当线程对象启动时最终会调用 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 {
        // 如果 task 不是 null 或者去 workQueue 队列中取到待执行的任务不为 null
        while (task != null || (task = getTask()) != null) {
            // ... ...
            try {
                // 开始执行任务前的钩子方法
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                    // ... ...
                } finally {
                    // 任务执行后的钩子方法
                    afterExecute(task, thrown);
                }
            } finally {
                // ... ...
            }
        }
        completedAbruptly = false;
    } finally {
        // Worker 退出
        processWorkerExit(w, completedAbruptly);
    }
}

runWorker 方法,首先会取出要执行的任务 task,如果为空则会调用 getTask 方法从任务队列中获取,然后调用任务对应的 run 方法进行执行,另外预置了 beforeExecute、afterExecute 两个钩子函数,让研发人员监控线程执行成为可能。

另外,线程池中的线程如何从队列中获取待执行的任务的呢?走进 getTask 方法看一看。

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    // 这块体现了:线程池的线程是复用的,通过循环去获取队列中的任务去执行。
    for (;;) {
        int c = ctl.get();
        // ... ...
        int wc = workerCountOf(c);
        // allowCoreThreadTimeOut: 是否允许核心线程超时.
        // 如果设置为 false,那么线程池在达到 corePoolSize 个工作线程之前,不会让闲置的工作线程退出。
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        // ... ...
        try {
            // 从 workQueue 队列中取待执行的任务,根据 timed 来选择等待时间
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

为了便于理解,源码做了部分删减。重点关注从任务队列中获取待执行任务的对象的方法调用:workQueue.poll()、workQueue.take() ,前者是移除并返回队列中的头部元素,如果队列为空,则返回 null,而后者是移除并返回队列中的头部元素,如果队列为空,则阻塞。

烟未灭,酒过半 ... ... 源码探讨就谈到这儿... ...

5

源码揭秘之后的反思

(一)钩子函数的使用场景

场景一:

如上面自定义的 MyThreadPoolExecutor,可以让日志打印线程及线程数等等信息。意味着研发人员可以扩展 ThreadPoolExecutor,对钩子函数 beforeExecute、afterExecute 进行实现,进而可以知晓线程池内部的调度细节,可以有效进行监控,针对故障排查应该很有帮助。

场景二:

AbstractExecutorService 并没有实现 execute 方法,而是为子类 ThreadPoolExecutor 留了个口子,让子类去灵活扩展(钩子函数)。

仔细想想业务开发时,诸多的使用场景,何尝不是如此呢?

(二)线程池的 submit 方法与 execute 方法啥区别呢?

execute 方法,适用于不需要关注返回值的场景,只需要将线程丢到线程池中去执行就可以了。

而 submit() 方法,适用于需要关注返回值的场景,不过最终会调用 execute() 方法。

考虑到性能提升,如果不需要关注返回值,则建议直接调用 execute() 方法,因为那样会屏蔽很多中间调度。

(三)线程池状态与线程数量用一个 ctl 变量表示的好处?

线程池状态和线程数量合二为一,用一个原子变量来表示,来减少锁竞争,提高并发效率。

(四)清晰的注释是否有必要?

通过探秘源码,很多图都是根据源码注释勾勒出来的。可以看出清晰的注释,对于核心流程而言真的很重要,一定要养成良好的编码习惯,关键业务逻辑、核心流程,建议一定要写好注释,利人又利己,何乐而不为之。

(五)Executor 家族框架,若写基础框架时,是否有借鉴意义呢?

个人感觉很有借鉴意义,因为无论业务开发还是基础服务,总会看到类似模式框架的身影,总会有大牛模仿着造轮子,所以闲暇之余可以抽象一下。

6

寄语写最后

本次,主要对 Executor 家族进行了简单介绍,并着重对线程池背后的 ThreadPoolExecutor 类进行深度剖析,知其然知其所以然,希望对大家有帮助。

本文分享自微信公众号 - 一猿小讲(yiyuanxiaojiangV5),作者:一猿小讲

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2020-08-05

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 揭秘字节跳动埋点数据实时动态处理引擎(附源码)

    宝贝们,还记得前几天博主去的火山引擎大数据场嘛,其中比较令大家感兴趣的就是最后一讲,字节一站式埋点平台的 flink 标准化清洗及拆流任务。

    公众号:大数据羊说
  • 使用线程池时候当程序结束时候记得调用shutdown关闭线程池

    日常开发中为了便于线程的有效复用,线程池是经常会被用的工具,然而线程池使用完后如果不调用shutdown会导致线程池资源一直不会被释放。下面通过简单例子来说明该...

    加多
  • Java线程池深入理解

    之前面试baba系时遇到一个相对简单的多线程编程题,即"3个线程循环输出ADC",自己答的并不是很好,深感内疚,决定更加仔细的学习《并发编程的艺术》一书,到达掌...

    用户1216676
  • 深入理解Java线程池

    ThreadPoolExecutor采取上述步骤的总体设计思路,是为了在执行execute()方法时,尽可能地避免获取全局锁(那将会是一个严重的可伸缩瓶颈)。在...

    luoxn28
  • 深入理解 Java 线程池

    线程池是一种多线程处理形式,处理过程中将任务添加到队列,然后在创建线程后自动启动这些任务。

    静默虚空
  • Java 线程池深入分析

    ScheduledThreadPoolExecutor 适用于延时执行,或者周期性执行的任务调度,ScheduledThreadPoolExecutor在实现上...

    纯洁的微笑
  • 深度解读Java线程池思想及实现

    相信很多Java从业者都听说过或者使用过Java的线程池,但真正了解线程池的并不多,这篇文章详细解读Java线程池设计思想及源码实现。线程池是一个非常重要的工具...

    迟恩
  • 源码分析-使用newFixedThreadPool线程池导致的内存飙升问题

    使用无界队列的线程池会导致内存飙升吗?面试官经常会问这个问题,本文将基于源码,去分析newFixedThreadPool线程池导致的内存飙升问题,希望能加深大家...

    捡田螺的小男孩
  • 创建线程以及线程池时候要指定与业务相关的名字,以便于追溯问题

    日常开发中当一个应用中需要创建多个线程或者线程池时候最好给每个线程或者线程池根据业务类型设置具体的名字,以便在出现问题时候方便进行定位,下面就通过实例来说明不设...

    加多
  • javaer的目标,是星辰大海!

    有一天,我去爬山。突然脚一滑,掉下了悬崖。摔的眼冒金星,精神破碎,最终掉到一个山洞中。山洞中央有一个宝盒,打开以后是一本书,细看是一份JAVA开发者的秘籍。举目...

    xjjdog
  • 成为一名Java高级架构师你应该要学习的技术

    Java天坑
  • 2019最新Java架构师学习路线图

    一个Java程序员不想要成为一个Java “old”程序员,唯一的出路就是学习Java技术的脚步永不停滞,让自己成为一个Java架构师,成为一个公司真正的技术大...

    动力节点Java培训
  • 揭秘深度强化学习

    用户1737318
  • 揭秘深度强化学习

    尽管监督式和非监督式学习的深度模型已经广泛被技术社区所采用,深度强化学习仍旧显得有些神秘。这篇文章将试图揭秘这项技术,并解释其背后的逻辑。受众读者主要是有机器学...

    CSDN技术头条
  • String没那么简单!

    String类相信大家都不陌生,我们写工程使用String的频率,就像写作文时使用汉字“的”的频率一样高。那么你经常使用它,是否真的“了解”它?请带着问题,一步...

    Java技术江湖
  • 搞定JVM基本原理和面试题,看看这几本书妥妥够了!

    在介绍本书单之前,我想先问一下各位读者,你们之前了解过JVM么,如果有了解过,那么了解的程度又如何呢。经过了10多年的发展,Java Web从开发框架到社区都已...

    程序员书单
  • 深入理解Java之线程池

    转载自:http://www.importnew.com/19011.html#comment-653957

    天涯泪小武
  • 2019 最新 Java 核心技术教程,都在这了!

    以下是Java技术栈微信公众号发布的所有关于 Java 的技术干货,会从以下几个方面汇总,本文会长期更新。

    Java技术栈
  • JVM(二)Java虚拟机组成详解

    Java虚拟机(Java Virtual Machine)下文简称jvm,上一篇我们对jvm有了大体的认识,进入本文之后我们将具体而详细的介绍jvm的方方面面,...

    Java中文社群-磊哥

扫码关注云+社区

领取腾讯云代金券