Java 线程池ThreadPoolExecutor（基于jdk1.8）（一）

介绍

线程池的作用就是提供一种对线程的管理，避免由于过多的创建和销毁线程所造成的开销。在一个“池”中维护着一定数量的线程，达到可重复利用的效果。在Java中，线程池的实现主要是通过ThreadPoolExecutor来实现的。接下来先从类图结构来分析一下。

类结构

Executor public interface Executor { void execute(Runnable command); }

在这看得出Executor是一个顶层的接口，里边只有execute方法。这个接口的目的就是表明需要执行一个Runnable任务。

ExecutorService 它直接继承至Executor，但是依然是一个接口，只是在此基础上增加了其他的方法。 <T> Future<T> submit(Callable<T> task);

submit方法主要就是接受Callable接口的参数，不再是Runnable参数了，而且增加了返回值Future。 在ExecutorService类中还有好几个重载函数。这几个方法的设计主要是为了让执行任务者能够得到任务的运行结果。

void shutdown();

这个方法主要是提供了关闭线程池的操作，调用此方法后，线程池不再接收新的任务，但是会把当前缓存队列的任务全部执行完毕。

List<Runnable> shutdownNow();

这个方法调用后，不但不能接收新的任务，也会尝试中断正在执行的任务，同时不再执行缓存队列中的任务。

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;

这个方法提供执行一系列的任务的功能，最后返回所有任务的Future对象，用于活动任务的执行结果。

AbstractExecutorService 它是一个抽象类，实现了ExecutorService接口。对其中绝大部分的方法进行了实现。 public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }

任务为空直接返回空指针异常，新建一个ftask对象，最终还是调用了execute方法去执行任务。现在追踪一下newTaskFor方法

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { return new FutureTask<T>(callable); }

返回的是一个FutureTask对象，这个类即实现了Runnable接口又实现了Callable接口，这样它既可以当作线程的执行对象又可以对任务执行后的结果进行获取。（为了不脱离主线，暂时就不再分析FutureTask的源码了）

public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException { if (tasks == null) throw new NullPointerException(); ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size()); boolean done = false; try { for (Callable<T> t : tasks) { RunnableFuture<T> f = newTaskFor(t); futures.add(f); execute(f); } for (int i = 0, size = futures.size(); i < size; i++) { Future<T> f = futures.get(i); if (!f.isDone()) { try { f.get(); } catch (CancellationException ignore) { } catch (ExecutionException ignore) { } } } done = true; return futures; } finally { if (!done) for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } }

首先创建用于存储结果的集合futures，大小为传入的任务数（一个任务对应一个future对象）。然后遍历所有的Callable对象，把它们封装到RunnableFuture中（实际传入的是futureTask对象），然后把创建的futureTask对象加入到结果集futures中，然后调用execute方法去依次执行传入的任务。 接下来又是一个for循环，在其中去保证每个任务已经执行完毕，当判断某一个任务if (!f.isDone())没有完成时，会调用f.get()，这个方法是一个阻塞方法，也就是说当前线程会一直等到任务执行结束才会返回。这样保证了所有的任务都会在这个for循环中全部执行完毕，然后返回futures结果集。 此抽象类中并没有实现execute、shutdown、shutdownNow等方法，具体的实现放在了ThreadPoolExecutor中。

ThreadPoolExecutor 这个类继承自AbstractExecutorService ，实现了execute方法。引入了线程池的管理。

源码分析

先看一下构造函数，因为线程池最重要的就是其中的一些参数，不同的参数配置，就可以实现不同的线程池。

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { ..... } corePoolSize 指的就是线程池的核心线程数。当当前线程池中的线程个数小于corePoolSize时，对新来的任务，直接开启一个新的线程去执行它。maximumPoolSize 代表最大能够容纳的线程数量。当线程池中的线程个数大于等于corePoolSize后，当需要执行一个新的任务时会先把任务放入缓存队列中，等待后续空闲的线程去执行。如果此时缓存队列已满，那么就会新启一个线程去执行它，如果线程数量已经超过了maximumPoolSize，那么就会调用reject方法，拒绝执行该次任务（后边会分析reject方法）。keepAliveTime 用于指定线程存活的时间，当线程池中的线程大于corePoolSize后，会监控每一个线程的空闲时间，如果某个线程的空闲时间大于keepAliveTime，那么就会销毁该线程，释放资源。

unit 这个是keepAliveTime的单位，可以为秒、毫秒等等。

workQueue 这个就是我们的任务缓存队列了。是一个阻塞队列的类型，常用的有ArrayBlockingQueue、LinkedBlockingQueue（默认容量是Integer.MAX_VALUE）和SynchronousQueue。

threadFactory 这个就是创建线程的工厂类。用于新建线程实体。

handler 这是拒绝某个任务的回调。当线程池不能够处理某个任务时，会通过调用handler.rejectedExecution()去处理。内置了四种策略 AbortPolicy（默认情况）：直接丢弃，并且抛出RejectedExecutionException异常。 DiscardPolicy：直接丢弃，不做任何处理。 DiscardOldestPolicy：从缓存队列丢弃最老的任务，然后调用execute立刻执行该任务。 CallerRunsPolicy：在调用者的当前线程去执行这个任务。

分析完了构造函数，想必大家对其中的参数有了一定的认识，可以看出，不同的参数组合就可以实现不同需求的线程池，当然Java中已经为我们内置了很多常用的线程池，它们全部位于Executors类当中。如果没有特殊需求，建议直接使用其中的线程池。 看完了构造函数，接下来分析最重要的execute函数，看看到底是怎么一个执行流程。英语好的直接阅读注释，已经写的非常仔细了。

public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * 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. */ int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } 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); } else if (!addWorker(command, false)) reject(command); }

不喜欢看文字的直接看下边的流程图：

既不想看注释也不喜欢流程图的，咱们直接分析代码: 首先是判断

workerCountOf(c) < corePoolSize

workerCountOf(c)其实就是计算当前线程池中的线程数，如果小于核心线程数那么直接

addWorker(command, true)

如果添加成功，那么方法就结束了。添加失败，则需要

isRunning(c) && workQueue.offer(command)

其实就是判断线程池是否处于运行状态，且尝试往缓存队列添加任务，如果同时为真，则：

int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false);

进入if的语句块，前边不是已经判断过是否处于运行状态了吗？为什么还要再次判断呢？因为考虑多线程访问，而且没有进行同步措施，所以需要再次进行double-check。 如果添加任务到缓存队列失败，那么就尝试添加新的线程：

else if (!addWorker(command, false)) reject(command);

如果添加失败就需要拒绝此次任务了。 次方法中关键调用的几个方法为：addWorker（），workQueue.offer()，reject()，接下来一个一个分析。