线程池

  JAVA5提供了多种类型的线程池,如果你对这些线程池的特点以及类型不太熟悉或者非常熟悉，请帮忙看看这篇文章(顺便帮忙解决里面存在的问题,谢谢！)：
    http://xtu-xiaoxin.iteye.com/admin/blogs/647580
   
    如果对ThreadPoolExecutor还不是很熟悉，可以看看一篇对ThreadPoolExecutor的介绍的博文：
http://blog.csdn.net/waterbig/archive/2009/11/10/4794244.aspx

    首先,JAVA中使用ThreadPoolExecutor的常用方式：
    实例代码1
Java代码 
Runnable runnable = new CountService(intArr);  
       ThreadPoolExecutor execute = (ThreadPoolExecutor)Executors.newFixedThreadPool(10);  
       //或者使用：ThreadPoolExecutor execute = (ThreadPoolExecutor)Executors.newCachedThreadPool();  
       execute.submit(runnable); 

Runnable runnable = new CountService(intArr);
       ThreadPoolExecutor execute = (ThreadPoolExecutor)Executors.newFixedThreadPool(10);
       //或者使用：ThreadPoolExecutor execute = (ThreadPoolExecutor)Executors.newCachedThreadPool();
       execute.submit(runnable);

    在分析ThreadPoolExecutor源码前，先了解下面两个概念：
     1.核心线程(任务)：我们定义的线程，即实现了Runnable接口的类，是我们将要放到线程池中执行的类，如实例代码中的CountService类
     2.工作线程：由线程池中创建的线程，是用来获得核心线程并执行核心线程的线程（比较拗口哦,具体看代码就知道是什么东东了）。

    Executors是一个线程池工厂，各种类型的线程池都是通过它来创建的，注意把它和Executor分开，感觉这个线程池工厂命名有点问题。
    我们主要分析下我们提交任务的处理逻辑，即’execute.submit(runnable)’的实现。
Submit()方法是在ThreadPoolExecutor继承的抽象类AbstractExecutorService中实现的，具体代码如下：
   Java代码 
public Future<?> submit(Runnable task) {  
        if (task == null) throw new NullPointerException();  
       //对核心线程的一个包装，RunnableFuture还是一个Runnable  
        RunnableFuture<Object> ftask = newTaskFor(task, null);  
       //核心线程执行逻辑  
        execute(ftask);  
        return ftask;  
    } 

public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
       //对核心线程的一个包装，RunnableFuture还是一个Runnable
        RunnableFuture<Object> ftask = newTaskFor(task, null);
       //核心线程执行逻辑
        execute(ftask);
        return ftask;
    }

    从代码中可以看出，线程的执行逻辑通过execute()完成，而execute是在AbstractExecutorService的子类ThreadPoolExecutor中实现的。看，一个典型的模板模式！废话少说，下面看ThreadPoolExecutor中execute()方法中代码：
  
    Java代码 
public void execute(Runnable command) {  
        if (command == null)  
            throw new NullPointerException();  
        /* 
         * command线程运行的整个逻辑在 addIfUnderCorePoolSize(command)方法中实现 
         * 一般适用于FixedThreadPool 
         */ 
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {  
            /* 
             * poolSize >= corePoolSize条件成立情景：当创建的为CacheThreadPool时,条件 
             * 就能成立 
             */ 
            if (runState == RUNNING && workQueue.offer(command)) {  
                if (runState != RUNNING || poolSize == 0)  
                    //两种情况下执行该方法:1.线程池shutdown  2.CacheThreadPool中第一个核心线程的执行  
                    ensureQueuedTaskHandled(command);  
            }  
            //CacheThreadPool中线程的执行逻辑  
            else if (!addIfUnderMaximumPoolSize(command))  
                reject(command); // is shutdown or saturated  
        }  
    } 

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * command线程运行的整个逻辑在 addIfUnderCorePoolSize(command)方法中实现
         * 一般适用于FixedThreadPool
         */
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
        /*
        * poolSize >= corePoolSize条件成立情景：当创建的为CacheThreadPool时,条件
        * 就能成立
        */
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                //两种情况下执行该方法:1.线程池shutdown  2.CacheThreadPool中第一个核心线程的执行
                    ensureQueuedTaskHandled(command);
            }
            //CacheThreadPool中线程的执行逻辑
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
    }

    注意：CachedThreadPool和FixedThreadPool的逻辑实现都是在ThreadPoolExecutor中实现的。它两的主要区别就是属性corePoolSize以及workQueue的初始值的不同。具体可自己查看工程类Executors的newFixedThreadPool()和newCachedThreadPool方法。由于这些初始值的不同，所以实现的逻辑也不同，具体的我在代码中已经注释了。
    command线程运行的整个逻辑在 addIfUnderCorePoolSize(command)方法中实现的，
详细请看addIfUnderCorePoolSize(command)源码：

  Java代码 
private boolean addIfUnderCorePoolSize(Runnable firstTask) {  
       Thread t = null;  
       final ReentrantLock mainLock = this.mainLock;  
       mainLock.lock();  
       try {  
        //poolSize < corePoolSize 即当前工作线程的数量一定要小于你设置的线程最大数量  
        //CachedThreadPool永远也不会进入该方法，因为它的corePoolSize初始为0  
           if (poolSize < corePoolSize && runState == RUNNING)  
               t = addThread(firstTask);  
       } finally {  
           mainLock.unlock();  
       }  
       if (t == null)  
           return false;  
       t.start();   //线程执行了  
       return true;  
   } 

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
        //poolSize < corePoolSize 即当前工作线程的数量一定要小于你设置的线程最大数量
        //CachedThreadPool永远也不会进入该方法，因为它的corePoolSize初始为0
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();   //线程执行了
        return true;
    }

    看’t.start()’，这表示工作线程启动了，工作线程t启动的前提条件是’t = addThread(firstTask); ‘返回值t必须不为null。好了，现在想看看java线程池中工作线程是怎么样的吗？请看addThread方法：
    Java代码 
private Thread addThread(Runnable firstTask) {  
    //Worker就是典型的工作线程,所以的核心线程都在工作线程中执行  
       Worker w = new Worker(firstTask);  
       //采用默认的线程工厂生产出一线程。注意就是设置一些线程的默认属性，如优先级、是否为后台线程等  
       Thread t = threadFactory.newThread(w);   
       if (t != null) {  
           w.thread = t;  
           workers.add(w);  
         //没生成一个工作线程 poolSize加1，但poolSize等于最大线程数corePoolSize时，则不能再生成工作线程  
           int nt = ++poolSize;    
           if (nt > largestPoolSize)  
               largestPoolSize = nt;  
       }  
       return t;  
   } 

private Thread addThread(Runnable firstTask) {
    //Worker就是典型的工作线程,所以的核心线程都在工作线程中执行
        Worker w = new Worker(firstTask);
        //采用默认的线程工厂生产出一线程。注意就是设置一些线程的默认属性，如优先级、是否为后台线程等
        Thread t = threadFactory.newThread(w);
        if (t != null) {
            w.thread = t;
            workers.add(w);
          //没生成一个工作线程 poolSize加1，但poolSize等于最大线程数corePoolSize时，则不能再生成工作线程
            int nt = ++poolSize; 
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }

   看见没，Worker就是工作线程类，它是ThreadPoolExecutor中的一个内部类。下面，我们主要分析Worker类，如了解了Worker类，那基本就了解了java线程池的整个原理了。不用怕，Worker类的逻辑很简单，它其实就是一个线程，实现了Runnable接口的，所以，我们先从run方法入手，run方法源码如下：

  Java代码 
public void run() {  
            try {  
                Runnable task = firstTask;  
                firstTask = null;  
                /** 
                 * 注意这段while循环的执行逻辑,没执行完一个核心线程后,就会去线程池 
                 * 队列中取下一个核心线程,如取出的核心线程为null,则当前工作线程终止 
                 */ 
                while (task != null || (task = getTask()) != null) {  
                    runTask(task);  //你所提交的核心线程(任务)的运行逻辑  
                    task = null;  
                }  
            } finally {  
                workerDone(this); // 当前工作线程退出  
            }  
        }  
    } 

public void run() {
            try {
                Runnable task = firstTask;
                firstTask = null;
                /**
                 * 注意这段while循环的执行逻辑,没执行完一个核心线程后,就会去线程池
                 * 队列中取下一个核心线程,如取出的核心线程为null,则当前工作线程终止
                 */
                while (task != null || (task = getTask()) != null) {
                    runTask(task);  //你所提交的核心线程(任务)的运行逻辑
                    task = null;
                }
            } finally {
                workerDone(this); // 当前工作线程退出
            }
        }
    }

    从源码中可看出，我们所提交的核心线程(任务)的逻辑是在Worker中的runTask()方法中实现的。这个方法很简单，自己可以打开看看。这里要注意一点，在runTask()方法中执行核心线程时是调用核心线程的run()方法，这是一个寻常方法的调用，千万别与线程的启动(start())混合了。这里还有一个比较重要的方法，那就是上述代码中while循环中的getTask()方法，它是一个从池队列中取的核心线程（任务）的方法。具体代码如下：

    Java代码 
Runnable getTask() {  
        for (;;) {  
            try {  
                int state = runState;  
                if (state > SHUTDOWN)    
                    return null;  
                Runnable r;  
                if (state == SHUTDOWN)  //帮助清空队列  
                    r = workQueue.poll();  
               /* 
                * 对于条件1，如果可以超时，则在等待keepAliveTime时间后，则返回一null对象,这时就 
                *  销毁该工作线程，这就是CachedThreadPool为什么能回收空闲线程的原因了。 
                * 注意以下几点:1.这种功能情况一般不可能在fixedThreadPool中出现 
                *            2.在使用CachedThreadPool时，条件1一般总是成立,因为CachedThreadPool的corePoolSize 
                *              初始为0 
                */ 
                else if (poolSize > corePoolSize || allowCoreThreadTimeOut)  //------------------条件1  
                    r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);    
                else 
                    r = workQueue.take();       //如果队列不存在任何元素 则一直等待。 FiexedThreadPool典型模式----------条件2  
                if (r != null)  
                    return r;  
                if (workerCanExit()) {       //--------------------------条件3  
                    if (runState >= SHUTDOWN) // Wake up others  
                        interruptIdleWorkers();  
                    return null;  
                }  
                // Else retry  
            } catch (InterruptedException ie) {  
                // On interruption, re-check runState  
            }  
        }  
    } 

Runnable getTask() {
        for (;;) {
            try {
                int state = runState;
                if (state > SHUTDOWN) 
                    return null;
                Runnable r;
                if (state == SHUTDOWN)  //帮助清空队列
                    r = workQueue.poll();
               /*
                * 对于条件1，如果可以超时，则在等待keepAliveTime时间后，则返回一null对象,这时就
                *  销毁该工作线程，这就是CachedThreadPool为什么能回收空闲线程的原因了。
                * 注意以下几点:1.这种功能情况一般不可能在fixedThreadPool中出现
                *            2.在使用CachedThreadPool时，条件1一般总是成立,因为CachedThreadPool的corePoolSize
                *              初始为0
                */
                else if (poolSize > corePoolSize || allowCoreThreadTimeOut)  //------------------条件1
                    r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS); 
                else
                    r = workQueue.take();       //如果队列不存在任何元素 则一直等待。 FiexedThreadPool典型模式----------条件2
                if (r != null)
                    return r;
                if (workerCanExit()) {       //--------------------------条件3
                    if (runState >= SHUTDOWN) // Wake up others
                        interruptIdleWorkers();
                    return null;
                }
                // Else retry
            } catch (InterruptedException ie) {
                // On interruption, re-check runState
            }
        }
    }

    从这个方法中，我们需要了解一下几点：
    1.CachedThreadPool获得任务逻辑是条件1，条件1的处理逻辑请看注释，CachedThreadPool执行条件1的原因是：CachedThreadPool的corePoolSize时刻为0。

    2.FixedThreadPool执行的逻辑为条件2,从’workQueue.take()’中我们就明白了为什么FixedThreadPool不会释放工作线程的原因了(除非你关闭线程池)。

    最后，我们了解下Worker(工作线程)终止时的处理吧，这个对理解CachedThreadPool有帮助，具体代码如下：
    Java代码 
/** 
    * 工作线程退出要处理的逻辑 
    * @param w 
    */ 
   void workerDone(Worker w) {  
       final ReentrantLock mainLock = this.mainLock;  
       mainLock.lock();  
       try {  
           completedTaskCount += w.completedTasks;   
           workers.remove(w);  //从工作线程缓存中删除  
           if (--poolSize == 0) //poolSize减一,这时其实又可以创建工作线程了  
               tryTerminate(); //尝试终止  
       } finally {  
           mainLock.unlock();  
       }  
   } 

/**
     * 工作线程退出要处理的逻辑
     * @param w
     */
    void workerDone(Worker w) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;
            workers.remove(w);  //从工作线程缓存中删除
            if (--poolSize == 0) //poolSize减一,这时其实又可以创建工作线程了
                tryTerminate(); //尝试终止
        } finally {
            mainLock.unlock();
        }
    }

    注意workDone()方法中的tyrTerminate()方法，它是你以后理解线程池中shuDown()以及CachedThreadPool原理的关键，具体代码如下：  

    Java代码 
private void tryTerminate() {  
    //终止的前提条件就是线程池里已经没有工作线程(Worker)了  
       if (poolSize == 0) {  
           int state = runState;  
           /** 
            * 如果当前已经没有了工作线程(Worker),但是线程队列里还有等待的线程任务,则创建一个 
            * 工作线程来执行线程队列中等待的任务 
            */ 
           if (state < STOP && !workQueue.isEmpty()) {      
               state = RUNNING; // disable termination check below  
               Thread t = addThread(null);  
               if (t != null)  
                   t.start();  
           }  
           //设置池状态为终止状态  
           if (state == STOP || state == SHUTDOWN) {  
               runState = TERMINATED;  
               termination.signalAll();   
               terminated();   
           }  
       }  
   }