并发基础（二） ThreadPoolExecutor

概述：

ThreadPoolExecutor -> AbstractExecutorService =>ExecutorService =>Executor.

ThreadPoolExecutor 解决了两个不一样的问题：1，经过减小任务调用开销来改善执行大量异步任务时的性能问题；2，提供资源约束与管理，包括线程。另外还有一些统计功能，好比已完成任务个数等。

为了使用于不一样的上下文环境，ThreadPoolExecutor提供了不少可调整的参数和钩子方法， 可是官方仍然推荐编程者使用更为便捷的 Executors 的工厂方法，这些方法根据不一样场景进行了配置，能够知足大部分需求：

 newCachedThreadPool() (无大小限制线程池，线程自动回收), 

newFixedThreadPool(int) (线程池大小固定) 

以及 newSingleThreadExecutor() (单个后台线程). 

 当你须要直接使用ThreadPoolExecutor 时， 可参考一下指南：

• Core and maximum pool sizes

   ThreadPoolExecutor 能够根据corePoolSize和maximumPoolSize自动调整线程池的大小(分别经过 getPoolSize() 和 getCorePoolSize()进行查看) . 当有新的任务提交时会调用execute(Runnable), 若是当前正在运行的线程数小于corePoolSize，新的线程会被建立来执行新提交的任务，即便当前有空闲线程。若是当前运行的线程数大于corePoolSize可是小于maximumPoolSize, 同时任务队列（queue）已经满了的话，一个新的线程会被建立来执行任务. 设置corePoolSize 和 maximumPoolSize 相等时, 至关于建立了一个固定大小的线程池. 若是设置 maximumPoolSize 为一个无限值好比Integer.MAX_VALUE, 那么该池中将容许有任意数量的并发任务. 通常状况下 corePoolSize和  maximumPoolSize 在构造时传入, 不过也能够经过 setCorePoolSize(int) 和 setMaximumPoolSize(int)来动态设定.

• On-demand construction

  默认状况下核心线程（core threads）只在有新任务到达时才会初始化和启动, 不过能够经过覆写 prestartCoreThread() 或者 prestartAllCoreThreads()方法来改变这种行为.

• Creating new threads

  默认经过 ThreadFactory建立新的线程. 它们将有相同的 ThreadGroup ，具备 NORM_PRIORITY优先级而且是非守护线程. 你能够提供不一样的ThreadFactory 来改变线程的名字、组别、优先级、守护状态等. 若是ThreadFactory 建立线程失败，该executor 将继续运行，可是不能再执行任何任务.

• Keep-alive times

  若是当前线程池中线程的数量已经多于corePoolSize, 那么额外的线程若是处于idle状态超过期间 keepAlive(查看 getKeepAliveTime(TimeUnit))将会被强行终止. 这样当线程池的使用比较紧张时能够有效地减小资源消耗.当线程池再次紧张时，新的（额外）线程还会被建立. keepAlive时间能够经过 setKeepAliveTime(long, TimeUnit)方法来设置 .使用Long.MAX_VALUE NANOSECONDS  可使处于idle状态的线程不被终止，直到ThreadPoolExecutor被关闭 （shut down）. 默认状况下， keep-alive 策略只在当前线程数量多于 corePoolSize 时起做用.可是能够经过方法 allowCoreThreadTimeOut(boolean) 设置是否将这样的超时策略同时应用于core Threads, 前提是 keepAliveTime 是非零的.

• Queuing

  任何 BlockingQueue 均可以用来调度和存储提交的任务. 任务队列的使用和线程池大小的改变有依赖关系:

• 若是当前线程数量小于coreThreadSize, Executor 会优先选择建立新线程来执行任务，而不是将任务存入队列.

• 若是有多于或等于corePoolSize个线程在运行, Executor 会将新的任务请求入队，而不是建立新的线程.

• 若是任务请求不能入队，新线程会被建立，可是要求线程数量不超过 maximumPoolSize, 不然任务请求会被拒绝(rejected).




队列有三种普遍使用的策略:

1. 直接传递(Direct handoffs). 一个不错的默认选择是使用同步队列(SynchronousQueue), 将任务直接交给线程处理而不是存储起来. 若是没有线程及时处理，任务也不能被成功存储，因此新的线程会被建立. 这种方法在处理一组有互相依赖的任务时避免了繁复的查找. 直接传递策略通常要求一个无界的线程池（unbounded maximumPoolSizes ），这样才能避免提交的任务被拒绝. 这种作法存在一个缺点，当任务到达速度超过线程对任务的处理速度是，线程数量会无限制的增加.

2. 无界队列(Unbounded queues). 若是当前的corePoolSize个线程都处于忙状态时，使用无界队列 (好比 LinkedBlockingQueue ,不预先设置容量) 会把新的任务缓存在队列里. 这样executor里不会有多于corePoolSize 个线程. ( 这样maximumPoolSize 就再也不起做用了.) 无界队列策略适用于任务相互独立的状况，各个任务不会影响彼此的执行， 好比 web page 服务器处理页面请求. 虽然这种方法可使爆发式请求获得平滑的处理，可是当任务处理速度较慢时，可能形成任务队列的无限制增加.

3. 有界对类(Bounded queues). 有界队列(好比 ArrayBlockingQueue) 是直接传递和无界队列的一个平衡，经过设置一个有限的maximumPoolSizes值，能够防止资源的过分消耗， 可是这种方式更难调整和控制. 任务队列大小和最大线程池大小须要互相折中: 使用较大的队列和较小的线程池能够减小CPU、系统资源 以及上下文切换开销, 可是会影响吞吐量. 当任务频繁阻塞时 (好比等待I/O),这种作法会对系统的调度形成负面影响 . 若是使用较小的任务队列就要求有较大的线程池, 这样可使CPU充分利用，可是会引起过分的调度开销，从而也会映像吞吐量.

• Rejected tasks

  若是executor已经关闭，或者executor使用了有界的线程池和有界的任务队列而且都已饱和，那么经过方法 execute(Runnable) 提交的新任务会被拒绝。 不管哪一种状况， execute 方法都会调用 RejectedExecutionHandler的 rejectedExecution(Runnable, ThreadPoolExecutor) 方法. RejectedExecutionHandler有四中预设的处理:

1. 默认使用 ThreadPoolExecutor.AbortPolicy, 当有任务被拒绝时会抛出运行时异常： RejectedExecutionException .

2. ThreadPoolExecutor.CallerRunsPolicy, 这种策略在调用 execute 方法的线程（即提交任务的线程）中运行任务 . 这种策略提供了一个简单的返回机制来使下降新任务的提交频率。

3. ThreadPoolExecutor.DiscardPolicy, 这种策略只是简单的丢弃新任务，不作任何反馈处理.

4. ThreadPoolExecutor.DiscardOldestPolicy, 若是 executor 还没有关闭, 任务队列首部的任务会被丢弃, 而后重试提交任务 (若是再次失败，还要重复这一过程.)

定义并使用其余的处理策略也是能够的. 可是要格外注意特定的工做场景下的队列容量和队列管理策略。

• Hook methods

  ThreadPoolExecutor提供了可覆写的beforeExecute(Thread, Runnable) 和 afterExecute(Runnable, Throwable)方法， 这些方法在每次执行任务前、后分别调用，可用来操做执行环境，好比从新初始化线程本地变量（ThreadLocals）,收集统计信息或者添加日志项 . 另外，还能够覆写 terminated() 方法在Executor彻底终止前作一些特殊的处理工做.

  若是钩子方法或回调方法抛出异常，内部的工做线程也可能会失败并异常终止。

• Queue maintenance

  方法 getQueue() 提供了访问queue的途径，从而对queue进行监视和调试，不推荐用于其余目的. 当有大量入队的任务被取消时，另外两个方法, remove(Runnable) and purge() 可用于存储资源的回收.

• Finalization

  线程池再也不被程序引用 而且 没有任何剩余的线程在运行时，线程池会被自动关闭 (shutdown ). 若是你想确保未被引用的线程池被回收， 即便用户忘记调用 shutdown(), 那么你必须使未被使用的线程最可以终止, 能够经过设置适当的 keep-alive 时间, 使用下限为0的核心线程数而且/或者  设置allowCoreThreadTimeOut(boolean)来实现.







下面看一下提交一个任务到执行的具体过程：

/*

          * Proceed in 3 steps:

          *

          * 1. If fewer than corePoolSize threads are running, try to

          * start a new thread with the given command as its first

          * task.  The call to addWorker atomically checks runState and

          * workerCount, and so prevents false alarms that would add

          * threads when it shouldn't, by returning false.

          *

          * 2. If a task can be successfully queued, then we still need

          * to double-check whether we should have added a thread

          * (because existing ones died since last checking) or that

          * the pool shut down since entry into this method. So we

          * recheck state and if necessary roll back the enqueuing if

          * stopped, or start a new thread if there are none.

          *

          * 3. If we cannot queue task, then we try to add a new

          * thread.  If it fails, we know we are shut down or saturated

          * and so reject the task.

          */

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        
        int c = ctl.get();
        //scene 1
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // scene 2
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // scene 3
        else if (!addWorker(command, false))
            reject(command);
    }

从注释里能够看到任务被添加时有三种状况，scene 1 和 scene 3 都容易理解， 咱们看第二种， addWorker(null, false):

addworker, 这里的firstTask 是null， 可是worker里的thread仍是启动了， 为何？

 private boolean addWorker(Runnable firstTask, boolean core) {
         
       ...
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    ...
                        workers.add(w);
                        ...
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

Worker也是一个runnable， 这里的线程启动运行的是Worker的run方法，里面又调用了runWorker 方法：

咱们看下 worker的构造：

 private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        ...
        
        /** Delegates main run loop to outer runWorker  */
        public void run() {
            runWorker(this);
        }

        ...
    }

下面的runWorker方法中注意while loop， 当前的worker若是有task就执行，不然调用getTask 从workQueue中取一个来执行，执行完会继续去取， 这里就体现了“线程池”的概念， 并非一个任务一个线程，而是在线程数量达到限定的数量，同时任务数比较多被放入缓存队列的时候， 一个线程有可能执行多个任务。

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                ...
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        
                        ...
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

看一下getTask方法：

这里有一个无限的for loop， 直到从workQueue取到一个任务，或者等待超时，或者当前的executor被关闭

 private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

从上面的过程能够看出， 经过addWorker(null, false /true)提交的任务虽然是null， 可是仍然会启动一个线程去workQueue中等待或取得一个任务。