Android 线程切换（3）: ThreadPoolExecutor 的定义和使用

参考

Java线程池之ThreadPoolExecutor
Java多线程线程池（4）–线程池的五种状态
《Android开发艺术探索》第11章：Android的线程和线程池——任玉刚
Java 多线程干货系列—（一）Java 多线程基础
线程的优先级与控制
深度解读 java 线程池设计思想及源码实现
Executors类中创建线程池的几种方法的分析
线程池的种类，区别和使用场景
JUC源码分析-集合篇（八）：SynchronousQueue
并发编程之线程池的使用及扩展和优化

关于ThreadPoolExecutor

出于管理线程执行过程及复用线程的目的，JDK使用了线程池ThreadPoolExecutor。ThreadPoolExecutor大体上，由一个存放线程的线程集（workers）、一个生成的线程的工厂类（threadFactory）和一个存放待执行任务的队列（workQueue）组成。其中线程工厂和任务队列都是可以由我们通过构造函数自定义的，线程集则是一个HashSet 。我们通过线程池，可以监听任务执行的开始、结束，以及中断执行队列中的任务。线程池通过阻塞等方法维持线程的存活，使得可以直接使用原有线程执行新的任务，节省了执行任务时创建销毁线程的开销。关于线程池的详细描述可以看这篇文章。

线程池的参数介绍

线程池的构造方法很多，最后都是调用到了 ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) 这个方法，其他的构造方法只是提供了相应的默认参数。这个方法的每个参数的含义如下：

• corePoolSize：核心线程池数量。
• maximumPoolSize：最大线程池数量，即核心线程和非核心线程的数量之和。
• keepAliveTime：非核心线程空闲时的存活时间。
• unit ： keepAliveTime参数的单位。
• workQueue：存放待执行任务的队列，要求是阻塞队列，且队列的长度就是可以存放的最大任务数量。
• threadFactory ： 创建线程的工厂类。
• handler：拒绝策略，当线程池无法处理新增任务（当线程数量已经达到最大数量且任务队列已满，或者线程池已停止运行）时，这时就会触发拒绝策略，将异常报给调用者。

线程池执行任务的逻辑是：

• 当前线程数小于核心线程数corePoolSize时，创建线程执行任务。
• 否则 ，如果任务队列workQueue未满，将任务放入任务队列。
• 否则，如果线程数小于最大允许线程数maximumPoolSize，创建线程执行任务。
• 否则，触发拒绝策略。

由方法参数可以猜出，线程池默认核心线程的存活时间是无限的，直至线程池停止才会停止运行。因此线程池还提供了一个方法allowCoreThreadTimeOut(boolean value)使核心线程与非核心线程一样受keepAliveTime限制，即线程池中所有线程在超过预定空闲时间之后都会自动停止。

默认参数

上一节提到了线程池有很多构造方法，就是因为ThreadPoolExecutor提供了对应的默认参数，这一节就来认识一下这些默认参数。首先先列举一下构造方法：

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue);public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)

最后一个即是参数最全的构造方法，对比可以看出corePoolSize、maximumPoolSize、keepAliveTime和unit、workQueue这几个参数是一定要我们自己定义的，而线程工厂threadFactory和拒绝策略handler则提供了默认值。

ThreadFactory

看源码知道，其中threadFactory的默认值是调用的Executors.defaultThreadFactory()，该方法返回值其定义DefaultThreadFactory如下:

private static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);private final ThreadGroup group;private final AtomicInteger threadNumber = new AtomicInteger(1);private final String namePrefix;DefaultThreadFactory() {
    SecurityManager s = System.getSecurityManager();group = (s != null) ? s.getThreadGroup() :Thread.currentThread().getThreadGroup();namePrefix = "pool-" +poolNumber.getAndIncrement() +"-thread-";}public Thread newThread(Runnable r) {
    Thread t = new Thread(group, r,namePrefix + threadNumber.getAndIncrement(),0);if (t.isDaemon())t.setDaemon(false);if (t.getPriority() != Thread.NORM_PRIORITY)t.setPriority(Thread.NORM_PRIORITY);return t;}}

newThread(Runnable r)方法即是ThreadFactory 接口提供的方法，线程池也是通过这个方法获取新线程的。在这个方法中，新的线程t 调用 setDaemon(boolean on)和setPriority(int newPriority)将新线程设为用户线程（保证线程不会应该随线程池所在线程的结束所结束导致任务没有执行完）和设置线程优先级（高优先级线程先执行的概率比低优先级的线程高，而非高优先级的线程就一定先比低优先级执行完）为默认优先级Thread.NORM_PRIORITY。

RejectedExecutionHandler

前文已经提到，拒绝策略用于当线程池无法处理新增任务时将异常报给调用者；RejectedExecutionHandler 的定义如下:

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

默认的拒绝策略为AbortPolicy:

public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() {
     }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    throw new RejectedExecutionException("Task " + r.toString() +" rejected from " +e.toString());}}

RejectedExecutionException是一个运行时异常，所以线程池默认的拒绝策略就是直接抛出异常。除此之外，ThreadPoolExecutor还定义了三个拒绝策略：CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy，这三个策略的代码如下：

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public CallerRunsPolicy() {
     }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
    r.run();}}}public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() {
     }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }}public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() {
     }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown()) {
    e.getQueue().poll();e.execute(r);}}}

可以看到，这三个策略在线程池状态停止状态（isShutdown方法涉及后面的内容，可以先这么理解）下，拒绝任务都是默认不处理的，除了这种情况之外，CallerRunsPolicy的处理方法是在调用线程池的线程（即调用execute方法的线程）执行拒绝任务；DiscardPolicy是直接忽略该任务；而DiscardOldestPolicy 则是将当前任务队列中的第一个任务抛掉，再加入原本被拒绝的任务，相当于每次要拒绝任务时都是拒绝等待时长最长的任务。
以上三种策略，加上默认的AbortPolicy ，ThreadPoolExecutor提供了四种拒绝策略，如果不满足需要，我们也可以实现RejectedExecutionHandler 创建自己的拒绝策略。

线程池的API

线程池提供了以下public方法（去除掉getter/setter方法）：

void execute(Runnable command)
void shutdown()
List<Runnable> shutdownNow()
boolean awaitTermination(long timeout, TimeUnit unit)
boolean prestartCoreThread()
int prestartAllCoreThreads()
boolean remove(Runnable task)
void purge()
//继承自AbstractExecutorService的方法
Future<?> submit(Runnable task)
Future<T> submit(Runnable task, T result)
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)

其中最重要的方法是execute 和shutdown，execute方法用来执行任务，submit和invoke*等方法最终调用的都是这个方法；shutdown用来停止线程池，它和shutdownNow方法的区别在于shotdown会等待所有任务执行完毕之后停止所有线程；而
shutdownNow不会执行任务队列中的任务。当然，shutdownNow不会中断正在执行的任务，因为shutdownNow调用的是Thread#interrupt方法，而这个方法不能中断线程（除非是在该线程内调用），只能写上标志，等待线程（执行完任务后）自己调用interrupt方法结束运行。

Executors 中的默认线程池

Executors 是一个工具类，里面提供了5种常用的线程池。

• newFixedThreadPool ： 固定数量线程的线程池。
• newCachedThreadPool：不限数量（最大数量Integer.MAX_VALUE）的可缓存线程池。
• newSingleThreadExecutor：只有一个线程的线程池。
• newScheduledThreadPool：不限数量，可重复或定时执行任务的线程池。
• newSingleThreadScheduledExecutor：单个线程，可重复或定时执行任务的线程池。

这几种线程大多是以特定参数来定制的线程池，具体如下（这些方法都还有一个可以自定义线程工厂的重载方法，实现类似，我就没有列出来了）：

newFixedThreadPool：

 public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}

由代码可以看出，newFixedThreadPool由nThreads个核心线程、0个非核心线程、无限长度（Integer.MAX_VALUE）的任务队列组成。适用于执行长期的任务。

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}

newCachedThreadPool由0个核心线程，不限数量（Integer.MAX_VALUE）的非核心线程，和一个同步队列组成。关于同步队列SynchronousQueue，具体实现可以看这篇文章，现在我们可以简单理解成SynchronousQueue是一个空队列，除非有另一个对象正好向这个队列读取元素（调用poll或take），此时这个元素就会从添加者直接转给读取者，否则向这个队列中添加元素时必然失败。在线程池的这个使用场景里，就意味着，当我们调用方法执行新的任务的时候，如果正有空闲线程（非核心线程）在从任务队列中取任务，就会由这个线程执行该任务，如果当前没有空闲的线程，就会新开一个线程执行该任务（具体逻辑下篇会谈到）。所以newCachedThreadPool适用于执行很多短期的异步的任务。

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
}

先不去管FinalizableDelegatedExecutorService是何许人也，它的参数就告诉了我们newSingleThreadExecutor的性质：这个线程池只有一个核心线程，和一个无限长的任务队列，所以它对任务执行顺序有要求的场景。

newScheduledThreadPool和newSingleThreadScheduledExecutor

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);}public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1));}public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue());}

这两个方法都是使用了ScheduledThreadPoolExecutor类，这个类是ThreadPoolExecutor的子类。它的作用是提供定时执行任务的功能。它的实现下篇再说，这两个线程显然是适用于周期性执行任务的场景。

设置线程池参数

以下是强迫症环节
如果是为了更有效控制工作线程的目的而使用的线程池，其实使用默认的线程池或者相应的参数就足够了。但是如何设置才能使得我们的线程池性能达到最好，资源利用率达到最高呢？经过我几个小时的百度和Google潜心研究，发现这个问题可以分成两个部分：CPU 和内存。先说结论——计算密集型（CPU利用高）的任务尽量使用较小的线程池，一般为CPU核心数+1；IO密集型任务 （读写文件、数据库）可以使用稍大的线程池，一般为2*CPU核心数。
理由有下：

• 对于利用CPU而言 ，可以参考知乎问题：java线程池大小为何会大多被设置成CPU核心数+1？和这篇文章第五节，线程数量要依据CPU核心数量而定，线程不该比CPU核心数量少。不然显然浪费了，但也不应该太多，因为切换线程也有消耗，差不多在CUP核心数量*2的时候就不会让CPU空置太多了。

• 线程太多还会导致内存溢出。参考这篇文章的实验。创建线程的CPU消耗虽然很小，但是jvm要为每个线程分配一定内存，所以当线程越来越多超过了线程堆栈的大小限制时，就会报错

  java.lang.OutOfMemoryError: pthread_create (1040KB stack) failed: Try again

  所以线程池还可以起到控制线程数量防止内存溢出的作用，这也是线程池不应该创建太多核心线程的原因。

以上说的线程数量，指的是核心线程和非核心线程的总和，那么核心线程和非核心线程的之间应该怎么分配呢？ 关于这个问题，还可以参考AsyncTask，这个类里也用到了线程池，它的线程池参数如下：

private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
// We want at least 2 threads and at most 4 threads in the core pool,
// preferring to have 1 less than the CPU count to avoid saturating
// the CPU with background work
private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
private static final int KEEP_ALIVE_SECONDS = 30;

注释中提过，AsyncTask将核心线程设置为2到4个，最好比CPU少一个，而最大线程数为CPU数量*2 +1，这也是符合我们前面得出的结论的，如果只是为了处理一般的情况，我们完全可以使用AsyncTask的这些参数来定义线程池。
顺带一提，从这里也可以看到，Java中获取CPU数量的方法是Runtime.getRuntime().availableProcessors()。