Java中的ThreadPoolExecutor线程池

摘要:今天简单来和大家一起学习一下,java中的ThreadPoolExecutor线程池。线程池简介背书中,线程池是一个并发框架,在初始化一个多线程应用程序过程中创立一个线程集合,而后在需要执行新的任务时重用这些线程而不是新建一个线程。正当使用线程池有以下几个好处1.线程池可以利用已经创立好的线程重复执

今天简单来和大家一起学习一下,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这个类
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开发手册中的一句话

阿里开发规约

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

  • 全部评论(0)
最新发布的资讯信息
【网页前端|HTML】最全面的前端开发指南(2019-07-15 12:57)
【系统环境|数据库】零基础如何快速学好大数据?(2019-06-29 12:27)
【系统环境|Linux】零基础如何学好大数据?必备需要学习知识(2019-06-18 11:54)
【系统环境|】Hadoop环境中管理大数据存储八大技巧(2019-06-15 11:01)
【系统环境|服务器应用】现在国内IT行业是不是程序员过多了?(2019-06-11 06:34)
【系统环境|服务器应用】新贵 Flutter(2) 自己设置 Widget(2019-06-11 06:34)
【系统环境|服务器应用】Android完整知识体系路线(菜鸟-资深-大牛必进之路)(2019-06-11 06:34)
【系统环境|服务器应用】Java程序员小伙经历三个月备战,终获阿里offer(2019-06-11 06:34)
【系统环境|服务器应用】每日一问:谈谈对 MeasureSpec 的了解(2019-06-11 06:34)
【系统环境|服务器应用】【科普】晶体管-1(2019-06-11 06:34)
手机二维码手机访问领取大礼包
返回顶部