Executor Framework分析 ThreadPoolExecutor主要参数分析

Posted ZhangJianIsAStark

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Executor Framework分析 ThreadPoolExecutor主要参数分析相关的知识,希望对你有一定的参考价值。

本篇博客主要记录ThreadPoolExecutor主要参数的含义,
并分析相关接口的具体实现。


很多时候,当我们不需要指定线程池的运行细节时,
会直接利用工具类Executors创建线程池,例如:

public class Executors 
    //创建固定大小的线程池
    public static ExecutorService newFixedThreadPool(int nThreads) 
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    

    ........

    //创建可复用的线程池
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) 
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);
    

    ......

从上述代码可以看出,Executors提供的方法,
实际上就是调用了ThreadPoolExecutor的构造函数,
只不过传入了不同的参数。

因此,我们有必要分析下ThreadPoolExecutor的构造函数,
了解下参数的具体含义,以便分析ThreadPoolExecutor的设计意图。

ThreadPoolExecutor的构造函数如下所示:

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) 
        ...................
    

接下来,我们来看看每个参数的含义。


一、corePoolSize和maximumPoolSize
corePoolSize和maximumPoolSize决定了线程池创建线程的方式和数量。

根据ThreadPoolExecutor的描述来看:
当新的任务被提交给线程池时,线程池会查看当前线程的数量。

如果当前线程的数量小于corePoolSize时,那么线程池就会创建出新的线程,
即使此时有空闲的线程存在。

如果当前线程的数量大于等于corePoolSize,且小于maximumPoolSize,
只有线程池的缓存队列满了时,才会创建新的线程。

为了验证这段描述,我们可以看看相关的源码:

public class ThreadPoolExecutor extends AbstractExecutorService 
    .........
    public void execute(Runnable command) 
        .......
        int c = ctl.get();
        //当前线程数小于corePoolSize时,
        if (workerCountOf(c) < corePoolSize) 
            //直接addWorker,增加线程,参数为true
            if (addWorker(command, true))
                return;
            c = ctl.get();
        

        //线程数大于corePoolSize时,先尝试将task加入队列workQueue
        if (isRunning(c) && workQueue.offer(command)) 
            ..............
        
        //加入队列失败后,即队列是满的,同样调用addWorker方法
        //参数为false
        else if (!addWorker(command, false))
            //创建失败,调用reject函数处理
            reject(command);
    

    ...............

我们跟进一下addWorker函数:

    private boolean addWorker(Runnable firstTask, boolean core) 
        retry:
        for (;;) 
            int c = ctl.get();
            ...........
            for (;;) 
                //得到当前线程数量
                int wc = workerCountOf(c);

                //若参数为true时,表示创建核心线程,总数必须小于corePoolSize
                //若参数为false时,表示创建非核心线程,总数必须小于maximumPoolSize
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;

                //增加线程数量,利用标号跳出循环
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                ...............
            
        

        //创建实际的worker,即工作线程
        ..............
    
    ................

从上面的代码可以看出,当ThreadPoolExecutor收到新的任务时,
即execute接口被调用后,才会调用addWorker函数创建线程。

如果想尽快地创建出核心线程,ThreadPoolExecutor提供了
prestartCoreThread和prestartAllCoreThreads函数,如下所示:

public class ThreadPoolExecutor extends AbstractExecutorService 
    ..........
    //创建一个core thread
    public boolean prestartCoreThread() 
        return workerCountOf(ctl.get()) < corePoolSize &&
            addWorker(null, true);
    
    ..........
    //创建出所有的core thread
    public int prestartAllCoreThreads() 
        int n = 0;
        while (addWorker(null, true))
            ++n;
        return n;
    
    ..........

二、keepAliveTime
当线程的数量超过corePoolSize后,为了减少对系统资源的消耗,
一旦线程池发现某个线程空闲的时间超过了keepAliveTime,
就会主动结束该空闲线程,以释放系统资源。

一般情况下,只有线程数量超过corePoolSize时,
才会试图回收空闲线程的资源(线程数量小于corePoolSize时,停止回收)。
不过,如果线程池调用了allowCoreThreadTimeOut接口,
则可以回收所有空闲的线程资源线程。

接下来,我们来看看对应的代码:

public class ThreadPoolExecutor extends AbstractExecutorService 
    ...............
    //我们仍然从addWorker入手
    private boolean addWorker(Runnable firstTask, boolean core) 
        .........
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try 
            //创建worker, 并得到其中的thread
            w = new Worker(firstTask);
            final Thread t = w.thread;

            if (t != null) 
                //判断线程池状态,决定是否能增加worker
                ..............
                if (workerAdded) 
                    //若增加了worker,则调用对应线程的start方法
                    t.start();
                    workerStarted = true;
                
            
         .........
        ...........
    
    ............

我们看看Worker类的实现:

    //Worker实现了Runnable接口
    private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    
        ..........
        Worker(Runnable firstTask) 
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;

            //调用getThreadFactory获取ThreadPoolExecutor对应的ThreadFactory
            //创建thread时,传入了worker作为runnable
            //因此,前文的线程start后,会调用worker的run方法
            this.thread = getThreadFactory().newThread(this);
        

        public void run() 
            //继续跟进runWorker函数
            runWorker(this);
        
        .........
    

跟进一下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 
            while (task != null || (task = getTask()) != null) 
                //不断获取task并执行
                ............
            
            completedAbruptly = false;
         finally 
            //没有task可执行时,就会移除worker,即回收对应的资源
            processWorkerExit(w, completedAbruptly);
        
    
    ............

从上面的代码可以看出,worker被创建后就会一直获取Task来执行。
如果获取不到task,worker就会被移除,对应的线程将会被回收。

最后,我们跟进一下getTask函数:

private Runnable getTask() 
    boolean timedOut = false;

    for (;;) 
        ...........
        int wc = workerCountOf(c);

        //在不调用allowCoreThreadTimeOut的条件下
        //线程池中线程数量大于corePoolSize时,就会判断生存时间
        //调用了allowCoreThreadTimeOut后,可将allowCoreThreadTimeOut置为true
        //所有线程都需要判断生存时间
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        //这里我们关注timedOut
        //后面的代码决定了它的值
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) 
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        

        try 
            //从workQueue中取出待执行的task
            //此处说明,当需要判断生存时间时,仅等待keepAliveTime
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            //没获取到task时,将timedOut置为true
            //于是下一轮for循环时,return null
            timedOut = true;
         catch (InterruptedException retry) 
            timedOut = false;
        
    

根据上面的代码应该不难理解keepAliveTime的含义。
线程池中的线程创建出来后,都是不断地尝试从同一个workQueue中获取任务。
keepAliveTime就是控制线程空闲时的生命周期。


三、WorkQueue
WorkQueue主要用于保存提交到线程池的任务。

根据第一部分提及的execute函数的源码,
我们可以得出以下结论:
1、当线程数量小于corePoolSize时,线程池会优先创建线程,
不会将任务加入到WorkQueue中;
2、当线程数大于corePoolSize时,线程池会优先将任务加入到WorkQueue中,
而不是创建新的线程;
3、当WorkQueue中任务数量达到上限时,线程池才会继续创建线程,
直到线程数达到maximumPoolSize;
4、如果线程数达到了maximumPoolSize,且WorkQueue中任务数量达到上限,
那么新来的任务将被丢弃掉。

常用的WorkQueue类型主要有以下三种:
3.1 无缓冲的队列
这种队列的代表是SynchronousQueue,其特点是:
收到任务后,会立即转交给工作线程处理,即队列本身不会缓存任务。

当线程池使用该类型队列时,如果有新任务到来,
但没有空闲线程,那么会立即尝试创建新的线程。
因此使用这种类型的队列时,为了避免任务被拒绝掉,
maximumPoolSizes一般被设置为Integer.MAX_VALUE。

3.2 无限缓存容量的队列
这种队列的代表是不设置容量上限的LinkedBlockingQueue(容量的上限比较大)。

当线程池使用该类型队列时,一旦线程的数量达到corePoolSize,
那么新到来的任务就会被加入到队列中。
于是,线程池创建线程的数量上限就是corePoolSize(maximumPoolSize没有意义)。

3.3 有限容量的队列
这种队列的代表是ArrayBlockingQueue。

当线程池使用该类型队列时,处理任务的行为,
与第三部分开头描述的完全一致。
即在队列满时,线程池就需要创建新的线程,
直到线程数量达到maximumPoolSizes。

容易看出,使用有限容量队列可以有效地节省系统资源。
不过,需要仔细平衡队列容量和maximumPoolSizes的值:
当队列容量较大,maximumPoolSizes较小时,虽然可以降低CPU使用率,
降低消耗的系统资源及线程切换带来的开销,不过此时工作的吞吐率将会降低;
相对地,当队列容量较小,就需要更大的maximumPoolSizes,
此时会提高CPU使用率,但增大线程切换的开销。


四、ThreadFactory
ThreadFactory负责创建实际的线程。

我们直接看Worker类相关的代码:

public class ThreadPoolExecutor extends AbstractExecutorService 
    ............
    private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    
        ..........
        Worker(Runnable firstTask) 
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            //调用getThreadFactory获取ThreadPoolExecutor对应的ThreadFactory
            this.thread = getThreadFactory().newThread(this);
        
        .........
    
    ............

利用工具类Executors创建ThreadPoolExecutor时,
如果不显示指定,一般默认使用DefaultThreadFactory。

public class Executors 
    ..........
    private static class DefaultThreadFactory implements ThreadFactory 
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        //决定了线程的group, 名称前缀
        DefaultThreadFactory() 
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        

        //创建实际线程的方法
        public Thread newThread(Runnable r) 
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);

            //强行定义了thread的daemon属性及优先级
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        
    
    ..........

可以根据需求,自行定义ThreadFactory。


五、RejectedExecutionHandler
RejectedExecutionHandler主要负责处理被拒绝的任务。

从第一部分execute的代码,我们知道当任务处理失败时,
将会调用reject方法:

public class ThreadPoolExecutor extends AbstractExecutorService 
    ..........
    final void reject(Runnable command) 
        //调用RejectedExecutionHandler的rejectedExecution方法
        handler.rejectedExecution(command, this);
    
    ..........

常见的RejectedExecutionHandler有以下几种:
5.1 AbortPolicy

public static class AbortPolicy implements RejectedExecutionHandler 
    ..........
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    

容易看出使用AbortPolicy时,一旦遇到任务无法处理的情况,将直接抛出异常。

5.2 CallerRunsPolicy

public static class CallerRunsPolicy implements RejectedExecutionHandler 
    ..........
    //调用线程池的execute方法时, 才有可能被拒绝,从而执行rejectedExecution方法
    //这是一个同步的过程,因此r.run将执行在调用者所在的线程中
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
        //调用前提是线程池没有被shutdown
        if (!e.isShutdown()) 
            r.run();
        
    

顾名思义,当使用CallerRunsPolicy时,一旦遇到任务无法处理的情况,
任务将直接在调用线程中执行。
因此使用CallerRunsPolicy时,最好不要在类似UI线程里直接执行execute函数,
避免UI线程被任务阻塞。

5.3 DiscardPolicy

public static class DiscardPolicy implements RejectedExecutionHandler 
    ............
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
        //直接丢弃
    

当使用DiscardPolicy时,一旦遇到任务无法处理的情况,线程池直接丢弃该任务。

5.4 DiscardOldestPolicy

public static class DiscardOldestPolicy implements RejectedExecutionHandler 
    ............
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
        if (!e.isShutdown()) 
            //移除workQueue中最前面的任务
            e.getQueue().poll();
            //重新调用execute,仍然有可能触发rejectedExecution
            e.execute(r);
        
    

当使用DiscardOldestPolicy时,一旦遇到任务无法处理的情况,
线程池直接将丢弃任务队列中最旧的任务,并重新调用execute接口处理任务。


六、总结
至此,我们已经分析完了ThreadPoolExecutor主要参数的含义。

结合参数的含义,分析ThreadPoolExecutor的execute函数,
应该比较容易掌握ThreadPoolExecutor的工作原理。

下一篇博客,我们再来分析ThreadPoolExecutor中其它值得一看的代码。

以上是关于Executor Framework分析 ThreadPoolExecutor主要参数分析的主要内容,如果未能解决你的问题,请参考以下文章

Executor Framework分析 ScheduledThreadPoolExecutor

Executor Framework分析 ScheduledThreadPoolExecutor

Executor Framework分析 ThreadPoolExecutor部分函数分析

Executor Framework分析 ThreadPoolExecutor部分函数分析

Executor Framework分析 ThreadPoolExecutor主要参数分析

Executor Framework分析 ThreadPoolExecutor主要参数分析