Java中的ThreadPoolExecutor线程池

今天简单来和大家一起学习一下,java中的ThreadPoolExecutor线程池。

线程池简介

背书中,线程池是一个并发框架,在初始化一个多线程应用程序过程中创建一个线程集合,然后在需要执行新的任务时重用这些线程而不是新建一个线程。

合理使用线程池有以下几个好处

1.线程池可以利用已经创建好的线程重复执行任务,避免了线程创建、销毁带来的系统资源上的开支。所以线程池减少了系统资源消耗
2.有任务时,线程池可以利用空闲的线程去执行任务,提高了系统的响应速率
3.使用线程可以提高系统的并发处理能力,但是不合理的创建、使用线程会严重损耗系统的资源,线程的上下文环境切换也会带来系统资源上的开支,使用线程池可以对线程进行统一的分配、调优、监控

线程池的实现原理

前面我已经在《并行执行任务的Fork/Join框架》一文中,给大家介绍了ForkJoinPool线程池,今天就以ThreadPoolExecutor为我们的猪脚吧,先举个栗子,然后在根据ThreadPoolExecutor使用,和大家一起分析源码。

举个栗子

/**
 * @Description: .
 * @Author: ZhaoWeiNan .
 * @CreatedTime: 2017/8/20 .
 * @Version: 1.0 .
 */
public class Demo {

    /**
     * 初始化线程池
     */
    private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 10, 20, TimeUnit.SECONDS, new ArrayBlockingQueue(10),
            new ThreadPoolExecutor.DiscardOldestPolicy());

    public static void main(String[] args) {
        Demo demo = new Demo();
        for (int i = 0; i < 20; i++) {
            final int time = i;
            //把线程添加到线程池中
            demo.threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println("任务索引是:" + time);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}
1.构造函数

通过构造函数来看看ThreadPoolExecutor的成员变量。

/**
     * 创建新线程的工厂。
     *
     */
    private volatile ThreadFactory threadFactory;

    /**
     * 当线程池中线程数量或者线程池停止时被调用的处理类.
     */
    private volatile RejectedExecutionHandler handler;

    /**
     * 空闲线程等待工作的超时时间
     */
    private volatile long keepAliveTime;

    /**
     * 默认为false: 核心线程即使在空闲的时候也保持存活
     *       true: 核心线程使用keepAliveTime去等待工作
     */
    private volatile boolean allowCoreThreadTimeOut;

    /**
     * 核心线程池的大小,保持存活的工作线程的最小数量。
     */
    private volatile int corePoolSize;

    /**
     * 线程池的最大容量
     */
    private volatile int maximumPoolSize;

    /**
     * 线程池中的任务队列,是用了阻塞队列来实现(有机会为大家介绍)
     */
    private final BlockingQueue<Runnable> workQueue;

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
                maximumPoolSize <= 0 ||
                maximumPoolSize < corePoolSize ||
                keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
2.看看把线程加入线程池的execute方法
 public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();

        /**
         * 按3个步骤走:
         */
        int c = ctl.get();
        /**
         * 1.如果少于核心线程池大小corePoolSize的线程处于RUNNING状态,
         * 开一个新的线程,把command对象作为该线程的第一个任务。
         * 调用addWorker方法,原子性操作判断command的状态和worker的数量
         * 防止在线程池不能添加线程的时候添加线程。
         */
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        /**
         * 2.如果一个任务成功加入到阻塞队列中,添加一个线程,还要进行第二次检查。
         * 因为一已经存在的线程可以能已经死亡,或者这个线程池已经关闭当要添加线程时。
         * 所以我们进行了第二次检查状态,进行必要的入队回滚,或者池中没有线程时去
         * 创建一个新的线程。
         */
        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);
        }
        /**
         * 3.如果任务不能入队,我们尝试创建一个新的线程,如果创建线程失败可,
         * 我们知道了,线程池现在是关闭状态或者是饱和状态,所以我们拒绝了这个任务。
         */
        else if (!addWorker(command, false))
            reject(command);
    }
3.看看addWorker方法
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 (;;) {
                //获取worker的数量
                int wc = workerCountOf(c);
                //判断worker的数量
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                //重新获取一次状态
                c = ctl.get();
                if (runStateOf(c) != rs)
                    continue retry;
                //由于worker数量改变导致CAS(原子操作)失败,重新进入内部循环
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        //顶一个一个worker对象
        Worker w = null;
        try {
            //创建一个可重入锁
            final ReentrantLock mainLock = this.mainLock;
            //利用传入的任务对象,初始化worker对象
            w = new Worker(firstTask);
            //创建一个对象
            final Thread t = w.thread;
            if (t != null) {
                //加锁
                mainLock.lock();
                try {
                    // 在获取到锁的情况下,再次检查状态
                    int c = ctl.get();
                    int rs = runStateOf(c);

                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        //先检查线程t是否已经start
                        if (t.isAlive())
                            throw new IllegalThreadStateException();
                        //加入到workers中
                        //看一眼 workers
                        // private final HashSet<Worker> workers = new HashSet<Worker>();
                        // workers保存了所有已经获取到了mainLock锁的worker对象
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
4.看看Worker这个类

《Java中的ThreadPoolExecutor线程池》 worker

Worker是ThreadPoolExecutor中的一个内部类,实现了Runnable接口,是一个线程类,主要使用来,执行加入到线程池中的任务

线程池的监控

线程池有一些属性,可以支持我们对线程池进行了一些监控,在出现问题的时候可以很方便的进行问题的定位:

    /**
     * 线程池里曾经创建过的最大线程数,如果该数等于线程池的大小,证明该线程池曾经被打满过
     */
    private int largestPoolSize;

    /**
     * 线程中已经完成的任务数量
     */
    private long completedTaskCount;

    /**
     * 获取线程池中活动的线程数
     */
    public int getActiveCount() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            int n = 0;
            for (ThreadPoolExecutor.Worker w : workers)
                if (w.isLocked())
                    ++n;
            return n;
        } finally {
            mainLock.unlock();
        }
    }

    /**
     * 线程池的线程数量
     */
    public int getPoolSize() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // Remove rare and surprising possibility of
            // isTerminated() && getPoolSize() > 0
            return runStateAtLeast(ctl.get(), TIDYING) ? 0
                    : workers.size();
        } finally {
            mainLock.unlock();
        }
    }

另外,线程池中,beforeExecute、afterExecute、terminated方法,都是空方法,可以重写这三个方法,在执行任务前、执行任务后、线程池关闭时执行代码来进行线程池的监控,例如可以统计任务执行时间等

    protected void beforeExecute(Thread t, Runnable r) { }

    protected void afterExecute(Runnable r, Throwable t) { }

    protected void terminated() { }

合理配置线程池

线程池的怎么配置,可能是一直困扰着我和大家的一个问题,具体合理配置我一直没有实践的做过实验,都是使用了度娘到的大神给的配置建议

CPU密集型任务

CPU密集型任务尽量配置少的线程数,一般为CPU个数 + 1

IO密集型任务

IO密集型任务线程并不是一直在执行任务,线程数可以设置多一点,一般为CPU个数的两倍

最后套用阿里巴巴Java开发手册中的一句话

《Java中的ThreadPoolExecutor线程池》 阿里开发规约

ThreadPoolExecutor就为大家简单的说到这,欢迎大家来交流,指出文中一些说错的地方,让我加深认识,愿大家没有bug,谢谢!

    原文作者:小草莓子桑
    原文地址: https://www.jianshu.com/p/ffda79c38f31
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞