管理你的线程池（Executor）

    我们都知道使用线程池能够控制线程的数量，尤其是大量的“短命”线程存在时，线程池将大大降低系统消耗（内存和CPU）。不过，线程池也同样需要管理，于是我写了本篇。
首先，我们来看看管理器的整个继承关系：


    显而易见，有ThreadPoolExecutor和ScheduledThreadPoolExecutor两个实现类，当然Executor类里也有一些内部类实现了特定的功能（如class DelegatedScheduledExecutorService），我们也可以自己通过扩展这里所有的接口、抽象类、类来实现自己的特定功能，如继承ThreadPoolExecutor类，覆写beforeExecute(),让它在每个任务开始执行前执行某些操作，还有很多可扩展功能，有兴趣的朋友可以自己摸索。
    你有两种方法创建上面管理器的实例：
1、你可以用上面介绍的两个类的那这些类的实例的构造函数来创建管理器的实例，不过你要自己配置一些诸如池最大尺寸（maximumPoolSize ）的参数。
2、Executors提供各种创建上面的类的实例的方法，它默认一些参数的设置。我主要介绍
这种方法中的newFixedThreadPool(int)和newCachedThreadPool()

------------newFixedThreadPool(int)------------
     创建一个默认尺寸的池，它同时运行的线程数将是固定的，如果你要让它课同时运行的最大线程数大于初始设置的那个参数，可以调用setMaximumPoolSize()来设置额外的线程来并行处理更多的任务。
     我们调用下面的方法来添加新的任务，到底Executors是如何处理的呢？

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if(poolSize>=corePoolSize|| !addIfUnderCorePoolSize(command)) {
//如果实时连接数小于corePoolSize，那么调用addIfUnderCorePoolSize()方法
            if (runState == RUNNING && workQueue.offer(command)) {
	//如果实时连接数大于了corePoolSize，那么将任务加进等待队列中。
                if (runState != RUNNING || poolSize == 0)
	//在执行workQueue.offer(command)的过程中shutdown了，确保所有的已经提交任务能够成功执行完。
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
	
                reject(command); // is shutdown or saturated
        }
    }

下面我们来看下poolSize>=corePoolSize为不同状态时两种执行方法：
 private boolean addIfUnderCorePoolSize(Runnable firstTask) {
	//首先获取本类所有同步方法的锁
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();
        return true;
    }
 
 private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
	//首先获取本类所有同步方法的锁
	 Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < maximumPoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();
        return true;

	}
几乎完全一样，估计author Doug Lea当初也是直接copy的吧。
这两个方法都调用了
private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);
	//这里并没有区分maximumPoolSize 和corePoolSize 
	Thread t = threadFactory.newThread(w);
        if (t != null) {
            w.thread = t;
            workers.add(w);//workers并没有尺寸的限制
            int nt = ++poolSize;
	//这一步维护一个管理器使用过程中的最大尺寸，没什么好说的。
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }
于是我认为发现管理器在对待aximumPoolSize 和corePoolSize 时根本没有什么区别，可是这是不正确的，至于为什么，大家可以自己去探索！

      ThreadPoolExecutor类内部有一个：
private final HashSet<Worker> workers = new HashSet<Worker>();
其中Worker类是ThreadPoolExecutor一个内部类，实现了Runable接口。在addIfUnderMaximumPoolSize()和addIfUnderCorePoolSize()两个方法中将任务添加进这个workers[]中，这个数组维护一个正在运行的任务组，这个数组中的一个元素对应一个正在运行的线程，如果一个线程以外死亡，数组中的元素没有被移除，管理器将自动创建一个新的线程继续从头开始执行刚刚那个以外死亡的数组对应的任务。
       如此神奇？那是如何实现的？
       很简单，ThreadPoolExecutor维护的线程的run方法都是在这个loop中的，

while (task != null || (task = getTask()) != null) {
                    runTask(task);
                    task = null;
                }

如果意外死亡，task=null不执行，重新判断条件的时候再次调用runTask(task);即，死亡的是runTask(task)方法内部的run()调用而已。

      说到这里，大家应该明白了，管理器无非就是用BlockingQueue<Runnable> workQueue队列（注意这个队列是线程安全的，挺有意思）来缓冲多出来的任务，而总是有不大于maximumPoolSize（注意，这里不是corePoolSize ）的线程在运行着，再有点异常死亡处理的能力而已。


--------newCachedThreadPool()--------

这个方法源码：
 public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
	}

    原来，就是让corePoolSize =0，maximumPoolSize=最大整数，然后设置空闲线程的存活时间为60s而已。看到这里，大家或许会冒出这样一个疑问：既然corePoolSize 是0，那么不是运行不了任何线程吗？呵呵，大家如果认真看了前面的文章就会有此疑问了。看同时刻运行的线程最大数是看参数maximumPoolSize不是corePoolSize 。
至于存活时间设置，那是很有必要，否则

while (task != null || (task = getTask()) != null) {
                    runTask(task);
                    task = null;
                }

getTask方法中从待执行任务缓冲队列中poll()任务的时候会有一个存活时间的超时机制，如果超时将返回null，这个线程将因为一系列连锁反应，最终死亡。

      好了，看似简单的Executor我砍了这么多，顺序整理的不是很好，大家将就看看吧。
总结一下，在设计这几个类的时候用到集合、同步（锁和阻塞队列）、枚举（TimeUnit）、多线程、安全控制（本文没有涉及）、工厂设计模式等等知识点，不简单哪^-^

评论

26 楼 angel243fly 2010-12-26  

没用过这些类呢，学习了

25 楼 C_J 2010-12-18  

挺好，再分析形象一点就更好了。
补充2点吧

引用

继承ThreadPoolExecutor类，覆写beforeExecute()

1，貌似不好extends，有很多final private，用组合来做更好。

2，ThreadPoolExecutor有4种中断策略来应对非常极端的没有线程可用且无法缓存任务的情况。

24 楼 贾懂凯 2010-12-17  

sam_kee 写道

能不能给个源码呢？O(∩_∩)O~，先谢楼主

估计这位仁兄没认真看，是JDK里的固有类。

23 楼 sam_kee 2010-12-17  

能不能给个源码呢？O(∩_∩)O~，先谢楼主

22 楼 贾懂凯 2010-12-17  

yunchow 写道

effective java  里讲的很清楚

除了effective java、Java编程思想，我推荐《Java 核心技术》（机械工业出版社，基础讲的很透彻）。
不知在Java核心技术方面大家还有什么好书推荐的？

21 楼 yunchow 2010-12-17  

effective java 里讲的很清楚

20 楼 贾懂凯 2010-12-16  

78425665 写道

贾懂凯 写道

78425665 写道

我有个问题，还请指教下。

像这样

		int i = 0;
		while(i++ < 50000){
			Object obj = "...";// 假设这个obj有点大，会占点内存，假如每个占1M
			threadPool.execute(new MyThread(obj));// threadPool继承的ThreadPoolExecutor
		}

假如corePoolSize=10，maximumPoolSize = 15吧

跑了不到一分钟，发现机器内存（1G）全被吃光了。队列都满了，可程序还在跑，threadPool还在加线程，这时它加到哪里去了？

建议测试一下，内存满了看是不是还能加。如果还能加，那只能说明你的内存计算方法是错的，根本没溢出，不是一个Object 1M10个Object就是10M，这种算法是错的。

没溢出，内存不会溢出，先不管我的内存计算方法

我是说，线程队列满了，然后循环程序还在跑，这时候线程池新加的线程，到哪里去了？

不是加的线程到哪里去的问题，而是如果你限制了任务缓存队列的尺寸，到底能不能加进去的问题！建议自己去看源码。
我可以给出一个结论-会抛出异常。

19 楼 78425665 2010-12-16  

贾懂凯 写道

78425665 写道

我有个问题，还请指教下。

像这样

		int i = 0;
		while(i++ < 50000){
			Object obj = "...";// 假设这个obj有点大，会占点内存，假如每个占1M
			threadPool.execute(new MyThread(obj));// threadPool继承的ThreadPoolExecutor
		}

假如corePoolSize=10，maximumPoolSize = 15吧

跑了不到一分钟，发现机器内存（1G）全被吃光了。队列都满了，可程序还在跑，threadPool还在加线程，这时它加到哪里去了？

建议测试一下，内存满了看是不是还能加。如果还能加，那只能说明你的内存计算方法是错的，根本没溢出，不是一个Object 1M10个Object就是10M，这种算法是错的。

没溢出，内存不会溢出，先不管我的内存计算方法

我是说，线程队列满了，然后循环程序还在跑，这时候线程池新加的线程，到哪里去了？

18 楼 贾懂凯 2010-12-16  

78425665 写道

我有个问题，还请指教下。

像这样

		int i = 0;
		while(i++ < 50000){
			Object obj = "...";// 假设这个obj有点大，会占点内存，假如每个占1M
			threadPool.execute(new MyThread(obj));// threadPool继承的ThreadPoolExecutor
		}

假如corePoolSize=10，maximumPoolSize = 15吧

跑了不到一分钟，发现机器内存（1G）全被吃光了。队列都满了，可程序还在跑，threadPool还在加线程，这时它加到哪里去了？

建议测试一下，内存满了看是不是还能加。如果还能加，那只能说明你的内存计算方法是错的，根本没溢出，不是一个Object 1M10个Object就是10M，这种算法是错的。

17 楼 78425665 2010-12-16  

我有个问题，还请指教下。

像这样

		int i = 0;
		while(i++ < 50000){
			Object obj = "...";// 假设这个obj有点大，会占点内存，假如每个占1M
			threadPool.execute(new MyThread(obj));// threadPool继承的ThreadPoolExecutor
		}

假如corePoolSize=10，maximumPoolSize = 15吧

跑了不到一分钟，发现机器内存（1G）全被吃光了。队列都满了，可程序还在跑，threadPool还在加线程，这时它加到哪里去了？

16 楼 hobitton 2010-12-16  

贾懂凯 写道

hobitton 写道

囧，为何题为管理线程池呢？这不是对着JDK源码讲了下线程池的实现方式吗？没搞懂。

http://blog.csdn.net/yangdengfeng2003/archive/2009/04/01/4042250.aspx

这篇和我说的一个主题，而且比我权威。不过，我特别不喜欢看那种结论性的文字，那令我头疼！从源码看清晰简单。

其实我主要是想问为啥取这个名字 

那篇文章和你的结合不就好了，看了源码总的需要全面总结下。每次记不得了还看下源码不是件很杯具的事情？

15 楼 yeshucheng 2010-12-16  

再深入下 就是BlockingQueue的原理，写的不错

14 楼 贾懂凯 2010-12-16  

hobitton 写道

囧，为何题为管理线程池呢？这不是对着JDK源码讲了下线程池的实现方式吗？没搞懂。

http://blog.csdn.net/yangdengfeng2003/archive/2009/04/01/4042250.aspx

这篇和我说的一个主题，而且比我权威。不过，我特别不喜欢看那种结论性的文字，那令我头疼！从源码看清晰简单。

13 楼 hobitton 2010-12-16  

囧，为何题为管理线程池呢？这不是对着JDK源码讲了下线程池的实现方式吗？没搞懂。

http://blog.csdn.net/yangdengfeng2003/archive/2009/04/01/4042250.aspx

12 楼 mib168 2010-12-16  

看来还得深入看，对线程池这块还比较陌生。

11 楼 zhaoxin1943 2010-12-16  

贾懂凯 写道

zhaoxin1943 写道

关于线程池，java编程思想里讲的比较好，个人觉得。

布置zhaoxin1943有没有相关文章，定拜读~~呵呵

本人小白加菜鸟，简称小白菜。看您的文章，受益匪浅。

10 楼 贾懂凯 2010-12-16  

zhaoxin1943 写道

关于线程池，java编程思想里讲的比较好，个人觉得。

布置zhaoxin1943有没有相关文章，定拜读~~呵呵

9 楼 zhaoxin1943 2010-12-16  

关于线程池，java编程思想里讲的比较好，个人觉得。

8 楼 peak 2010-12-15  

楼主最好把源代码贴出来研究一下，jdk写的这个线程池确实不错，以前都是我们自己写，现在用这现成的确实方便了很多

7 楼 贾懂凯 2010-12-15  

javamonkey 写道

maximumPoolSize 跟queue总类关系很大，如果是容量无限的话，maximumPoolSize 设置无效

刚看了一下确实如此，如果超过corePoolsize话，新的任务会首先尝试添加进等待队列queue中，如果添加进队列失败（超时或队列设置了固定的大小下溢出），才会利用maximunPoolSize属性。