并发编程 — 详解线程池

本文章需要对JDK5 Executor框架有所了解,请读者先阅读
初解线程池:http://ray-yui.iteye.com/blog/2072463


并发编程系列文章:
      初解线程池:http://ray-yui.iteye.com/blog/2072463
      详解线程池:http://ray-yui.iteye.com/blog/2075311
      并发数据类型:http://ray-yui.iteye.com/blog/2080454
      并发数据结构:http://ray-yui.iteye.com/blog/2084202
      volatile:http://ray-yui.iteye.com/blog/2231474
      判断线程安全:http://ray-yui.iteye.com/blog/2232931
      实现线程安全:http://ray-yui.iteye.com/blog/2234688


      本文将讲述如何通过JDK提供的API自定义定制的线程池

    //固定线程数 -- FixedThreadPool
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
    }

    //单条线程  --  SingleThreadExecutor
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));
    }

    //线程数自增长    --  CachedThreadPool
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>());
    }

      从上面代码可以看出,通过Executors创建的三种线程池其实内部都是调用
ThreadPoolExecutor进行创建

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

corePoolSize	指定线程池中线程的数量
maximumPoolSize	指定线程池中最大线程数量
keepAliveTime	空闲线程存活时间,指超过corePoolSize的线程
unit	keepAliveTime的时间单位
workQueue	任务队列,被提交但未被执行的任务存放容器
threadFactory	线程创建工厂,请参考上一章
handler	拒绝策略,当任务太多来不及处理时的拒绝策略

      需要重点注意的是workQueue和handler两个参数


workQueue:
      参数workQueue是指被提交但未被执行的任务所存放的队列容器,它是一个BlockingQueue接口,仅用于存放Runaable对象.根据队列功能的分类,在ThreadPoolExecutor中可以使用两种队列

直接提交队列:
      SynchronousQueue,SynchronousQueue是没有容量的容器,每一个插入的操作都需要等待相应的删除操作,SynchronousQueue不保存任务,它总是马上将任务提交给线程执行,如果没有空闲的线程则会尝试创建新的线程,如果线程数量已经达到最大值,则执行拒绝策略,使用SynchronousQueue通常需要设置很大的maximumPoolSize


有界的任务队列:
      有界队列可以使用ArrayBlockingQueue,ArrayBlockingQueue的构造函数必须传入一个容量参数,表示队列的最大容量,当使用有界队列并有新任务时,若然线程池线程数量小于corePoolSize则会创建现场,若然大于corePoolSize则会将新任务加入任务队列,当任务队列已满无法加入时,则在总线程数不大于maximumPoolSize的前提下创建线程,若大于maximumPoolSize则执行拒绝策略,使用有界队列除非系统非常繁忙,否则确保核心线程数在corePoolSize


无界的任务队列:
      无界任务队列可以使用LinkedBlockingQueue,与有界队列相比,除非系统资源耗尽,否则不会存在任务入队失败的情况.若任务创建和处理速度差异很大,无界队列会快速膨胀导致系统资源耗尽


优先任务队列:
      优先任务队列使用PriorityBlockingQueue实现,PriorityBlockingQueue是一个特殊的无界队列,创建PriorityBlockingQueue时可以传入Comparator对任务进行优先级处理,PriorityBlockingQueue和无界队列可能会发生的问题一样,不过PriorityBlockingQueue能控制任务的优先级别


handler
      handler参数指定了拒绝策略,即当任务数量超过线程池实际负载的时候,该如何处理被提交的任务,JDK内置提供了4种拒绝策略

AbortPolicy策略:
      该策略会直接抛出异常

CallerRunsPolicy策略:
      只要线程池未关闭,该策略直接在调用者线程中运行当前被放弃任务

DiscardOledestPolicy策略:
      该策略丢弃最老的一个请求,即即将被执行的任务,并尝试再提交当前任务

DiscardPolicy策略:
      该策略丢弃无法处理的任务,不做任何处理


     所有拒绝策略都继承自RejectedExecutionHandler接口,读者可根据实际情况需要扩展该接口实现自己的拒绝策略



扩展线程池
      ThreadPoolExecutor也是一个可扩展的线程池,它提供了beforeExecute,afterExecute,terminated3个接口对线程池调用任务进行控制

public class TestThreadPoolExecutor extends ThreadPoolExecutor {

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
    }

    @Override
    protected void terminated() {
        super.terminated();
    }
}

总结:
      Executor框架提供了多功能的定制方式让开发者快速方便的实现定制的线程池,从而减少不必要的开发工作,且本身也提供多个扩展点让开发者自行扩展实现独特的业务逻辑

评论

2 楼 mahongming 2014-06-27  

无界队列有误导人的嫌疑，看来大家都是相互复制的，LinkedBlockingQueue本身是可以设置长度的，如果不设置长度默认是int最大值,还有SynchronousQueue是缓冲区为1的队列，不是没有容量，是左手交右手的关系，既然是队列怎么可能没有容量呢

 public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

1 楼 dongya1987 2014-06-27  

很喜欢博主的排版