引言

本篇博客介绍通过“执行任务”的机制来设计应用程序时需要掌握的一些知识。所有的内容均提炼自《Java并发编程实战》中第六章的内容。

大多数并发应用程序都是围绕“任务执行”来构造的：任务通常是一些抽象的且离散的工作单元。

当围绕“任务执行”来设计应用程序结构时，第一步，就是要找出清晰的任务边界。在理想情况下，各个任务之间是相互独立的：任务并不依赖于其他任务的状态、结果或边界效应。

大多数服务器应用程序提供了一种自然的任务边界选择方式：以独立的客户请求为边界。

不好的例子：串行与“为每个任务创建线程”

在书中6.1节，介绍了由最简单的串行执行任务到为每个任务创建一个线程这两种执行任务的方式。应该说这两种方式都是不可取的。

这一节主要是为了引出下一节介绍的“任务执行框架”。

其中“串行执行任务”的缺点是在一般的服务器应用程序中，无法提高吞吐率或快速响应性。

而“为每个任务创建线程”的方式的问题在于可能导致：高性能开销、高资源消耗、影响稳定性。

【重点】在工作或面试中也会遇到这个极富针对性的问题，即大量创建线程会存在哪些问题？

1、高性能开销：创建和销毁都需要一定的代价，创建过程需要时间，延迟处理请求，也需要jvm和操作系统提供一些辅助操作。

2、高资源消耗：活跃的线程会消耗系统资源，尤其是内存。当可运行的线程数量多余可用处理器的数量，那么会有大量空闲的线程占用内存，不仅给垃圾回收带来压力，在竞争CPU的时候还将产生额外的性能开销。

3、影响稳定性：大量线程占用内存，内存不足，导致可能抛出OutOfMemoryError，系统崩溃。

线程数量的限制

书中在这里简单引出一个概念：稳定性。

根据前后文的联系，这里具体指的是：应用程序不会因为线程过多而抛出OutOfMemoryError异常。

为了达到这种稳定性，在可创建线程数量上存在一个限制。这个限制受平台以及多个因素影响，包括JVM启动参数、Thread构造函数中请求的栈大小、底层操作系统对线程的限制等。例如，在32位机器上，其中一个主要的限制因素是线程栈的地址空间。每个线程都维护两个执行栈，一个用于Java代码，另一个用于原生代码。

通常，JVM在默认情况下会生成一个复合栈，大小约0.5M~1M（这个值可以通过JVM标志 -Xss或通过Thread的构造函数来修改），那么：线程数量 ≈ 2^32(bit) / 0.5(MB) ≈几千或几万。

因此，在一定范围内，增加线程可以提高系统的吞吐率，但如果超出这个范围，再创建更多的线程只会降低程序的执行速度。

Executor接口

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

Executor是一个非常简单的接口，只有一个execute(Runnable) 方法，它是其他的灵活且强大的异步任务框架的基础。通过这种方式，用Runnable来表示任务，可以将任务的提交过程与执行过程解耦。

Executor本身就是基于生产者消费者，提交任务相当于生产者，执行任务相当于消费者，因此，如果要在程序中实现一个生产者-消费者的设计，那么最简单的方式通常就是使用Executor。

什么是执行策略？

执行策略，定义了任务执行的“what、where、when、how”等方面，主要是描述根据不同的资源而选择不同的执行方式，一个最优执行策略应当是与硬件资源最匹配的。

线程池

先来看一下四种常用线程池的创建：

ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);  
ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();  
ScheduledExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(10);  
ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();

其中：ExecutorService extends Executor，ScheduledExecutorService extends ExecutorService 。

1、newFixedThreadPool(int) ：创建一个定额线程池，每提交一个任务创建一个线程，达到数量限制后不再增加，这时线程池的规模将不再变化（如果某个线程由于发生了未预期的异常而结束，那么线程池会补充一个新的线程）

2、NewCachedThreadPool() : 创建一个可缓存的线程池，线程池的规模不存在任何限制，当线程多余任务时，回收空闲线程；当任务增加时，创建新线程。

3、NewSingleThreadExecutor：单线程的Executor，如果这个线程异常结束，会创建另一个线程来替代。NewSingleThreadExecutor能确保依照任务在队列中的顺序串行执行（例如FIFO、LIFO、优先级）。

4、NewScheduleThreadPool：创建一个固定长度的线程池，而且以延迟或定时的方式来执行任务，类似于Timer。

Executor的生命周期

JVM只有在所有（非守护）线程全部终止后才会退出，无法正确地关闭Executor，JVM将无法结束。

Executor以异步的方式来执行任务，导致了提交任务的状态不是立即可见的，即有些任务可能已经完成，有些可能正在执行，还有些可能正在队列中等待执行。

ExecutorSevice接口就是为了解决执行服务的生命周期问题，扩展了Executor接口。它添加了一些用于声明周期管理的方法（同时还有一些用于任务提交的便利方法）：

public interface ExecutorService extends Executor {  void shutdown();  List<Runnable> shutdownNow();  boolean isShutdown();  boolean isTerminated();  boolean awaitTermination(long timeout, TimeUnit unit)  throws InterruptedException;  // ......其他用于任务提交的便利方法  
}

这五个方法是声明周期管理的方法，其余的都是与任务提交相关的方法，比如，可以提交比较大的集合Callable对象的方法：

invokeAll(Collection<? extends Callable<T>> tasks)

【重点】ExecutorService的三种状态：运行、关闭、已终止。

ExecutorService在初始创建时处于运行状态。shutdown()方法将执行平缓的关闭过程：不再接受新的任务，同时等待已经提交的任务执行完成——包括那些还未开始执行的任务。shutdownNow()方法将执行粗暴的关闭方式：它将尝试取消所有运行中的任务，并且不再启动队列中尚未开始执行的任务。

延迟任务与周期任务

Timer类负责管理延迟任务以及周期任务，但它本身存在缺陷，因此通常要用ScheduleThreadPoolExecutor的构造函数或newScheduleThreadPool工厂方法来创建该类对象。

Timer的缺陷在于，Timer在执行所有定时任务时只会创建一个线程。如果某个任务的执行时间过长，那么将破坏其他TimerTask的定时精确性。

Timer还有一个问题就是，Timer线程不会捕获异常，当TimerTask抛出未检查异常时将终止定时线程。Timer也不会恢复线程的执行，而是会错误地任务整个Timer都被取消了。这就造成：已经被调度但尚未执行的TimerTask将不会再执行，新的任务也不会被调度。称之为“线程泄漏”。

【重点】生命周期小结

Runnable和Callable等任务的生命周期：创建、提交、开始、完成、取消。

Future表示的就是一个任务的生命周期。

Thread的生命周期：创建、就绪、运行、阻塞、死亡（或结束）。

ExecutorService的生命周期（因为它继承自Executor，因此也是Executor的生命周期）：创建、运行、关闭、已终止。

Callable与Future

callable

Runnable有一个局限性是没有返回值，也没办法抛出受检异常。对于某些异步获得结果的任务无法胜任，Callable应运而生。

它是Runnable的升级版，既可以使用Callable<Void>来达到Runnable一样的效果，同时也可以使用Callable<T> 来指定返回结果。

创建Callable的方式有两种：构造函数、静态的封装方法。

Callable<String> callableTask = new Callable<String>() {  @Override  public String call() throws Exception {  return "this is a callable task....";  }  
};

Java 8 style：

Callable<String> callableTask = () -> {  return "this is a callable task....";  
};

静态方法：Executors.callable(Runnable task, T result)：

Callable<String> call = Executors.callable(() -> {System.out.println("this is a runnable task...");
}, "done!");

Future

future表示一个任务的生命周期。主要提供了一些方法用于判断任务处于哪个阶段，还可以获取任务的结果甚至是取消任务。它本身还有一层隐含意义是，任务的生命周期只能前进，不能后退，当一个任务处于“完成”状态，就永远停留在“完成”状态上。这一点和ExecutorService的生命周期一样。

Future接口：

public interface Future<V> {  boolean cancel(boolean mayInterruptIfRunning);  boolean isCancelled();  boolean isDone();  V get() throws InterruptedException, ExecutionException;  V get(long timeout, TimeUnit unit)  throws InterruptedException, ExecutionException, TimeoutException;  
}

创建Future的方式通常是使用ExecutorService的submit()方法获取返回值。如果想通过构造器的方式显式地创建一个任务的生命周期管理对象，可以使用FutureTask。

FutureTask<String> runnFutureTask = new FutureTask<String>(runnable, "done!");  
FutureTask<String> callFutureTask = new FutureTask<>(callable);

FutureTask类实现了Runnable和Future两个接口。

（说明：FutureTask是Java 5加入的类，Java 6又为它补充了一个新的RunnableFuture接口，Runnable接口和Future接口被提升到了RunnableFuture接口上，这更像是一种重构手段，我个人认为在实际开发中用途可能不及直接使用FutureTask）

由于FutureTask实现了Runnable接口，因此可以将它提交给Executor来执行，或者直接调用它的run方法。

是的，FutureTask的run()方法可以直接执行任务，而不需要什么start。

Future.get

get()方法的行为取决于任务的状态（尚未开始、正在运行、已完成）如果任务已经完成，那么get会立即返回或抛出一个Exception；如果任务没有完成，那么get将阻塞直到任务完成。如果任务抛出异常，那么get将该异常封装成ExecutionException并重新抛出，可以通过getCause来进一步获得被封装的初始异常。如果任务被取消，那么get将抛出CancellationException。

异构任务并行化存在的局限

A与B两个完全不同的任务通过并行方式可以实现小幅度的性能提升，但是如果想大幅度的提升存在一定的困难。因此，得出一个结论是，只有当大量相互独立且同构的任务可以并发进行处理时，才能体现出真正的性能提升。

CompletionService与它的子类ExecutorCompletionService

CompletionService是Executor与BlockingQueue的融合。

回顾一下BlockingQueue的一些特性：

BlockingQueue接口是Queue的子接口，有两个最主要的实现，LinkedBlockingQueue（无界队列）和ArrayBlockingQueue（有界队列）。take()或poll()方法都是BlockingQueue的取头元素的方法，唯一不同的是当没有可用的头元素时，take会无限期等待（阻塞），poll可以设置一个超时时间，一旦超时，将返回null。

CompletionService是在任务执行的功能上加入了队列的特性，很明显是用于处理一批允许有返回值的任务。

用法：创建一个CompletionService（ExecutorCompletionService对象）。【ExecutorCompletionService的构造器允许我们传入一个ExecutorService（用于采取不同的执行策略）和一个BlockingQueue（该参数可选，默认LinkedBlockingQueue）】然后可以将一组Callable任务提交给CompletionService来执行，然后使用类似队列操作的take或poll方法来获取已完成的结果，这些结果会在完成时被封装为Future。

【扩展】ExecutorCompletionService的实现很简单。首先通过构造函数创建一个BlockingQueue来保存计算结果，然后当计算完成时，调用FutureTask的done方法，放入队列。展开：当提交某个任务时，该任务将首先包装为一个QueueingFuture，这是FutureTask【回顾：FutureTask实现了Future、Runnable】的一个子类，QueueingFuture改写了FutureTask的done方法——将结果放入BlockingQueue中。take和poll方法委托给BlockingQueue方法，这些方法会在得到结果之前阻塞。

为任务设置时限

有时候，如果某个任务无法在指定时间内完成，那么将不再需要它的结果，此时可以放弃这个任务。Future.get中支持这种需求：当结果可用时，它将立即返回，如果在指定时限内没有计算出结果，那么抛出TimeoutException。

在使用时限任务时需要注意，当这些任务超市后应该立即停止，从而避免为继续计算一个不再使用的结果而浪费计算资源。

【使用Future.get为单个任务设置时限，如果希望对一组任务设置计算时限，比如前面介绍的CompletionService，那么可以使用poll方法来设置执行时间】

invokeAll方法

ExecutorServie接口中有两个重载的invokeAll方法：

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)  throws InterruptedException;  
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,  long timeout, TimeUnit unit)  throws InterruptedException;

invokeAll方法支持将多个任务提交到一个ExecutorService并获得结果。invokeAll方法的参数为一组任务，并返回一组Future。invokeAll按照任务集合中迭代器的顺序将所有的Future添加到返回的集合中，从而使调用者能够将各个Future与其表示的Callable关联起来。

当所有任务都执行完毕时，或者调用线程被中断时，又或者超过指定时限时，invokeAll都会返回。当超过指定时限，任何还未完成的任务都会取消。当invokeAll返回后，每个任务要么正常完成，要么被取消，而客户端代码可以调用get或isCancelled来判断究竟是何种情况。