首页 > 代码库 > ThreadPoolExecutor源码学习（1）-- 主要思路
ThreadPoolExecutor源码学习（1）-- 主要思路
ThreadPoolExecutor是JDK自带的并发包对于线程池的实现，从JDK1.5开始，直至我所阅读的1.6与1.7的并发包代码，从代码注释上看，均出自Doug Lea之手，从代码上看JDK1.7几乎是重写了ThreadPoolExecutor的实现代码，JDK1.6的实现比较晦涩难懂，不便于理清作者的思路，故从JDK1.7的代码说起。
ThreadPoolExecutor既然是一个线程池，那么所谓池子的概念和意义多半在于充分利用资源，线程池当然就是为了充分利用线程资源，即尽量减少新建和销毁线程所产生的开销，带着这个问题，来看下作者是如何实现这个目的的。
典型的ThreadPoolExecutor调用
1 public ThreadPoolExecutor(int corePoolSize,2 int maximumPoolSize,3 long keepAliveTime,4 TimeUnit unit,5 BlockingQueue<Runnable> workQueue);//constructor6 7 public void execute(Runnable command);
典型的使用通常为new一个ThreadPoolExecutor对象，然后调用execute方法把需要执行的代码丢进去，剩下的线程控制就交给ThreadPoolExecutor管理了。从构造函数的corePoolSize，maxnumPoolSize都好理解，但是对于keepAliveTime，我之前使用一直存在一个无解，认为是execute线程所能运行的最大时间，实际上：
/** * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating.*/
当当前线程数量大于corePoolsize的时候，多出来的线程在没有任务执行的时候所能存在的最大时间。
ThreadPoolExecutor总体的代码量还是比较大，但是核心代码（对传入线程的处理，即execute方法）还是比较清晰的（JDK1.7）。整体思路如下：
1，如果线程数量小于corePoolsize则新建一个worker线程，执行传入的Runnable对象。
2，如果线程数量大于等于corePoolsize，则把当前传入的Runnable对象放到队列里面（workQueue，见构造函数）。
3，如果队列已满，则新建一个worker线程，来执行当前传入的Runnable对象。
4，如果队列已满，线程数量也达到maxnumPoolsize,则只能放弃执行传入的Runnable对象。
代码如下：
1 public void execute(Runnable command) { 2 if (command == null) 3 throw new NullPointerException(); 4 if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) { 5 if (runState == RUNNING && workQueue.offer(command)) { 6 if (runState != RUNNING || poolSize == 0) 7 ensureQueuedTaskHandled(command); 8 } 9 else if (!addIfUnderMaximumPoolSize(command))10 reject(command); // is shutdown or saturated11 }12 }
这里存在一个问题，那就是队列workQueue的对象在什么时候被执行呢？再仔细搜索代码的执行流程，每次新建的Thread并非只是把传入的Runnable对象直接执行，而是：
1 /** 2 * Main run loop 3 */ 4 public void run() { 5 try { 6 Runnable task = firstTask;//这里的first一般为传入的Runnable 7 firstTask = null; 8 while (task != null || (task = getTask()) != null) { 9 runTask(task);10 task = null;11 }12 } finally {13 workerDone(this);14 }15 }
这个为worker线程的run方法，其中的getTask()就是去workQueue里面取出Runnable对象并不停的执行，只有当workerQueue为空了，这个worker线程才退出这个死循环，结束这个线程。
以上为ThreadPoolExecutor整体实现思路，但是为了保证线程安全，作者也是下了不少功夫。选取比较有趣的几个地方说下：
对于保存通过execute方法传入进来的Runnable对象，作者保存在一个BlockingQueue里面，BlockingQueue是一个interface，里面有一个特性
/* * <p><tt>BlockingQueue</tt> methods come in four forms, with different ways * of handling operations that cannot be satisfied immediately, but may be * satisfied at some point in the future: * one throws an exception, the second returns a special value (either * <tt>null</tt> or <tt>false</tt>, depending on the operation), the third * blocks the current thread indefinitely until the operation can succeed, * and the fourth blocks for only a given maximum time limit before giving * up. */
即是对于获取一个可能会不能马上获取的对象，这个队列必须实现4种方式：1，抛异常，2，马上返回空，3，阻塞当前调用改方法的线程，4，阻塞该线程等待一定时间，那么从队列两个主要的方法来说，insert,remove就会有8个方法。如果我们使用new CachedThreadPool()，即一个常用的生成线程池的工厂方法，workQueue将是一个SynchronousQueue对象，这个对象的特点就是：
/** * A {@linkplain BlockingQueue blocking queue} in which each insert * operation must wait for a corresponding remove operation by another * thread, and vice versa. A synchronous queue does not have any * internal capacity, not even a capacity of one. You cannot * <tt>peek</tt> at a synchronous queue because an element is only * present when you try to remove it; you cannot insert an element * (using any method) unless another thread is trying to remove it; * you cannot iterate as there is nothing to iterate. The * <em>head</em> of the queue is the element that the first queued * inserting thread is trying to add to the queue; if there is no such * queued thread then no element is available for removal and * <tt>poll()</tt> will return <tt>null</tt>. For purposes of other * <tt>Collection</tt> methods (for example <tt>contains</tt>), a * <tt>SynchronousQueue</tt> acts as an empty collection. This queue * does not permit <tt>null</tt> elements.*/
这个队列的特点就是为空，必须在有线程想从里面获取对象并已经阻塞住了，才能往里面插入对象。如果线程池使用这样的队列实现，将会有什么效果呢？还是从 从队列里面获取Runnable对象的方法来看：（该方法为公共方法，无论传入什么队列）
Runnable getTask() { for (;;) { try { int state = runState; if (state > SHUTDOWN) return null; Runnable r; if (state == SHUTDOWN) // Help drain queue r = workQueue.poll();//2，马上返回 else if (poolSize > corePoolSize || allowCoreThreadTimeOut) r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);//等到一定时间，不行则放弃 else r = workQueue.take();//3，阻塞 if (r != null) return r; if (workerCanExit()) { if (runState >= SHUTDOWN) // Wake up others interruptIdleWorkers(); return null; } // Else retry } catch (InterruptedException ie) { // On interruption, re-check runState } } }
事实上，对于传入SynchronousQueue，最终只会调用阻塞的take()，对于调用poll()，会一直返回null，因为改队列不保存对象。那new CachedThreadPool()将会返回一个这样的线程池，如果线程的数目够用，那当前存在的线程将被重用（因take()一直在那里等待新的线程），如果当前的线程们执行不过来，则会新开线程，但是这个新开的线程不会主动销毁。
作者在对操作主要对象使用了ReentrantLock这样一个锁来实现对资源访问的同步除了如下这些优点以外，我觉得在代码清晰性上也是一个大的优势：
final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (poolSize < corePoolSize && runState == RUNNING) t = addThread(firstTask); } finally { mainLock.unlock(); }synchronized (lockObject) { if (poolSize < corePoolSize && runState == RUNNING) t = addThread(firstTask);}
明显第一种写法在语义上看更直观一点:-D
ThreadPoolExecutor源码学习（1）-- 主要思路