<返回更多

Java线程池深度揭秘

2020-08-12    
加入收藏

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

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

1. Executor 框架家族简介;

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

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

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

5. 源码揭秘之后的反思;

6. 寄语。

Executor 家族简介

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

Java线程池深度揭秘

executor 家族简图

(一)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 类的源码进行剖析。

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

Java线程池深度揭秘

 

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

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

Java线程池深度揭秘

 

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

/**
 * 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,可谓是设计甚妙!

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

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

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

Java线程池深度揭秘

 

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

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

Java线程池深度揭秘

 

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

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,而后者是移除并返回队列中的头部元素,如果队列为空,则阻塞。

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

源码揭秘之后的反思

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

场景一:

Java线程池深度揭秘

 

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

场景二:

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

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

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

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

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

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

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

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

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

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

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

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

寄语写最后

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

声明:本站部分内容来自互联网,如有版权侵犯或其他问题请与我们联系,我们将立即删除或处理。
▍相关推荐
更多资讯 >>>