ThreadPoolExecutor 源码分析（二）--- 队列的应用

一、序言

       上一篇，介绍了线程池的基本工作原理，这里会介绍一些里面的一些饱和策略和几个常用的线程池方法的实现原理。

 

二、源码分析

       线程池极力推荐我们用Executors 提供车的各种工厂，来创建我们的线程池，提供了我们常用的几种创建线程池的方式：

      newCachedThreadPool()：无界线程池，可以自己进行回收。

newFixedThreadPool(int)：固定线程池大小

      newSingleThreadExecutor()：单个线程

   

      我们来看看里面是如何实现的。

      

// 无界队列
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
// 这里我们看看这个构造里面做了什么
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
corePoolSize = 0 ;核心线程为0
maximumPoolSize： Integer 的最大值
keepAliveTime ： 60L,unit  单位是秒
workQueue ：SynchronousQueue 阻塞队列
Executors.defaultThreadFactory() ：默认工厂，也就是 DefaultThreadFactory
defaultHandler ：默认饱和策略，也就是 AbortPolicy 

 

   下面大概对这里面进行介绍：

    2.1 我们可以看出，核心线程数默认为0，最大是int范围最大值，也就是说在没超过这个范围的情况下，所有的线程都是会竞争执行，并且每个线程都只能维持keepAliveTime 时间，这里可以参考前一篇的源码解释，或者跟着源码run.

    2.2 SynchronousQueue ：单一的阻塞队列，也就是说无论我们是取元素(take)，或者存放元素(put)，都会阻塞，比如：我们存放了一个对象，那么它就会等待另一个线程将这个对象取出来，才能存放另一个，否则就一直阻塞，同理，如果我们取一个对象，如果里面没有，那么就一直阻塞到另一个线程，存放进去为止。是一个单一的生产消费模式。这里我们暂时不对它进行详细解释.

    2.3 DefaultThreadFactory ：这玩意儿提供默认创建线程的方式，具体看源码吧！这里这哪是不介绍。

    2.4 defaultHandler ：这里是默认的AbortPolicy 饱和策略，达到最大线程范围，就抛出异常了。

 

    可能上面的解释不够清晰，下面我们再对几个构造，里面使用的队列以及饱和策略进行介绍：

 

 3 。newFixedThreadPool(int) 固定线程池大小：

      

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


 // 可以看出，除了制定了核心线程数，以及LinkedBlockingQueue 以外，其他都一样
  public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

 

   3.1 该方法指定了corePoolSize 和 maximumPoolSize的数目，也就是说能保留最低线程数在线程池，而且超过了的线程数只能保存在队列里面，FIFO策略。

  3.2 LinkedBlockingQueue ：这是基于链表形式的队列，用于存放超出部分的线程。这是一个无界队列。

 

4.      newSingleThreadExecutor()：这里是 上面的翻版，仅仅是指定的数目为1而已。

 

 5.  可能大家比较好奇，为啥一个用SynchronousQueue 一个用 LinkedBlockingQueue ？

         我的理解是：

         5.1 SynchronousQueue  是不能存放多余的线程，也就是说如果我需要10个线程，当第一个开始执行，即使没执行完成，那么后续的线程加入也会从新创建，并执行，并不会加入队列里面。

         5.2 LinkedBlockingQueue 是可以存放线程的队列。如果我用newFixedThreadPool，核心线程和最大线程都控制在5个，当前面5个线程为执行完成之前，后面的线程只能被加入队列，不会创建，也就是说必须要等到前面线程执行完成了，才能从队列里面取出执行。

 

        从上面的解释可以看出这个到底怎么使用，当然还有有界队列（ArrayBlockingQueue），优先级队列（PriorityBlockingQueue）等队列，关于这些队列的内部实现，可以参考以前写的关于缓存的淘汰策略。

           关于饱和策略的实现，其实也差不多，这里也不暂时不进行源码分析，我copy 了一段介绍，相信能很清楚的解释：

         ThreadPoolExecutor.AbortPolicy：表示拒绝任务并抛出异常

           ThreadPoolExecutor.DiscardPolicy：表示拒绝任务但不做任何动作

           ThreadPoolExecutor.CallerRunsPolicy：表示拒绝任务，并在调用者的线程中直接执行该任务

           ThreadPoolExecutor.DiscardOldestPolicy：表示先丢弃任务队列中的第一个任务，然后把这个任务加进队列。

 

小结：

        1.这里本想说明一下线程池的细节，但是大多数围绕了几个队列的知识，里面的实现很多是靠队列本身的实现进行的。

        2.关于J.C.U 的一些队列源码的分析和应用，以后分析吧，最近比较忙，分析有问题的地方还请指出，相互学习。 

 

 

 

评论

2 楼 josico 2014-04-24  

楼主写得不错，对queue和handle都解释得很详细

1 楼 肥啦A梦 2014-04-09  

增长知识