线程池的一些疑问和解答

疑问

1. 线程池中的线程是如何实现一个线程执行多个任务的？

2. 构造线程池时为什么要用阻塞队列做为参数，非阻塞队列不行吗？

线程池的几个重要参数

• corePoolSize 是线程池的核心线程数，一般线程池会维持这个线程数

• maximumPoolSize 是线程池所能维持的最大线程数

• keepAliveTime 和 unit 则分别是超额线程的空闲存活时间数和时间单位

• workQueue 是提交任务到线程池的入队列

• threadFactory 是线程池建立新线程的线程构造器

• handler 是当线程池不能接受提交任务的时候的处理策略

线程池处理任务通常流程

1. 若是当前线程池中的线程数目小于corePoolSize，则每来一个任务，就会建立一个线程去执行这个任务；

2. 若是当前线程池中的线程数目>=corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列当中，若添加成功，则该任务会等待空闲线程将其取出去执行；若添加失败（通常来讲是任务缓存队列已满），则会尝试建立新的线程去执行这个任务；

3. 若是当前线程池中的线程数目达到maximumPoolSize，则会采起任务拒绝策略进行处理；

4. 若是线程池中的线程数量大于 corePoolSize时，若是某线程空闲时间超过keepAliveTime，线程将被终止，直至线程池中的线程数目不大于corePoolSize；若是容许为核心池中的线程设置存活时间，那么核心池中的线程空闲时间超过keepAliveTime，线程也会被终止。

分析

addIfUnderPoolSize

以最简单的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);        //建立线程去执行firstTask任务   
    } finally {
    mainLock.unlock();
}if (t == null)    return false;
t.start();return true;
}

addThread

看见了熟悉的t.start();，可是有一个问题,任务执行完线程就应该被销毁才对，为何线程池里的线程可以作到执行多个任务呢？ t = addThread(firstTask); 里确定大有名堂。

private Thread addThread(Runnable firstTask) {
Worker w = new Worker(firstTask);
Thread t = threadFactory.newThread(w);  //建立一个线程，执行任务   if (t != null) {
    w.thread = t;            //将建立的线程的引用赋值为w的成员变量       
    workers.add(w);    int nt = ++poolSize;     //当前线程数加1       
    if (nt > largestPoolSize)
        largestPoolSize = nt;
}return t;
}

Worker

上面的代码中出现了Worker这个类，来看看Worker的结构：

private final class Worker implements Runnable {
    private final ReentrantLock runLock = new ReentrantLock();
    private Runnable firstTask;
    volatile long completedTasks;
    Thread thread;
    Worker(Runnable firstTask) {
        this.firstTask = firstTask;
    }
    boolean isActive() {
        return runLock.isLocked();
    }
    void interruptIfIdle() {
        final ReentrantLock runLock = this.runLock;
        if (runLock.tryLock()) {
            try {
        if (thread != Thread.currentThread())
        thread.interrupt();
            } finally {
                runLock.unlock();
            }
        }
    }
    void interruptNow() {
        thread.interrupt();
    }
 
    private void runTask(Runnable task) {
        final ReentrantLock runLock = this.runLock;
        runLock.lock();
        try {
            if (runState < STOP &&
                Thread.interrupted() &&
                runState >= STOP)
            boolean ran = false;
            beforeExecute(thread, task);   //beforeExecute方法是ThreadPoolExecutor类的一个方法，没有具体实现，用户能够根据
            //本身须要重载这个方法和后面的afterExecute方法来进行一些统计信息，好比某个任务的执行时间等           
            try {
                task.run();
                ran = true;
                afterExecute(task, null);
                ++completedTasks;
            } catch (RuntimeException ex) {
                if (!ran)
                    afterExecute(task, ex);
                throw ex;
            }
        } finally {
            runLock.unlock();
        }
    }
 
    public void run() {
        try {
            Runnable task = firstTask;
            firstTask = null;
            while (task != null || (task = getTask()) != null) {
                runTask(task);
                task = null;
            }
        } finally {
            workerDone(this);   //当任务队列中没有任务时，进行清理工做       
        }
    }
}

关于Worker这个类，有几点须要注意：

1. Worker实现了runnable接口，能做为实例化thread的参数存在，thread.start()后执行的实际上是Worker的run()方法

2. 在Worker的run()方法中，有一段代码很是关键

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

   task!=nulll好理解，task = getTask()) != null又是什么意思呢？不罗嗦，继续跟代码

Runnable getTask() {
    for (;;) {
        try {
            int state = runState;
            if (state > SHUTDOWN)
                return null;
            Runnable r;
            if (state == SHUTDOWN)  // Help drain queue
                r = workQueue.poll();
            else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //若是线程数大于核心池大小或者容许为核心池线程设置空闲时间，
                //则经过poll取任务，若等待必定的时间取不到任务，则返回null
                r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            else
                r = workQueue.take();
            if (r != null)
                return r;
            if (workerCanExit()) {    //若是没取到任务，即r为null，则判断当前的worker是否能够退出
                if (runState >= SHUTDOWN) // Wake up others
                    interruptIdleWorkers();   //中断处于空闲状态的worker
                return null;
            }
            // Else retry
        } catch (InterruptedException ie) {
            // On interruption, re-check runState
        }
    }
}

workQueue

workQueue是BlockingQueue的一个实例，而BlockingQueue的take()方法当队列为空时会使当前线程发生阻塞
getTask()的功能就是去缓存队列获取任务，若是缓存队列为空，则处于等待状态，不然拿到task回去执行runTask()方法，在runTask()方法里执行咱们丢给线程池的任务的run()方法完成任务【兜了这么一大圈子】
经过while循环使这个过程不断进行

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

总结

其实把整个流程梳理一遍就是，将要执行的任务封装进Worker对象中，Worker实现Runnable接口重写run()方法实现不断获取新任务（无论是直接得到仍是来自缓存队列）的逻辑，将Worker做为参数实例化Thread对象，这样在开启线程后Worker的run方法就被间接调用了。至此，第一个问题就解决了。至于第二个问题，若是熟悉生产者消费者模式的话也不难理解。

参考资料：

• Java并发编程：线程池的使用

• ThreadPoolExecutor的应用和实现分析