转载

从0学习java并发编程实战-读书笔记-结构化并发应用程序（6）

在线程中执行任务

在理想情况下，任务之间都是相互独立的：任务并不依赖于其他任务的状态，结果或边界效应。独立性有助于实现并发。

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

串行地执行任务

最简单的方式就是在单个线程中串行的执行各项任务。但是现实中的web服务器的情况却并非如此。在web请求中包含了一组不同的运算和I/O操作。服务器必须处理套接字I/O以读取请求和写回响应，这些操作通常会由于网络或者连通性问题而被阻塞。

在服务器应用中，串行机制通常无法提供高吞吐率和快速响应性。

显示地为任务创建线程

如果为每个请求创建一个新的线程来提供服务，特点有：

任务处理过程将主线程中分离出来，使主循环能够更快的接受下一个到来的连接，使得程序在完成前面的请求之前可以接受更多新的请求，从而提高响应性。
任务可以并行处理，从而可以同时服务多个请求。程序的吞吐量将会提高。
任务处理代码必须是线程安全的，因为将会有多个任务并发的调用这段代码。

只要请求的到达速率不超过服务器的请求处理能力，那么这种方法可以同时带来更快的响应性和更高的吞吐率。

无限制创造线程的不足

在生产环境中，如果为每个任务分配一个线程，这种方法有着一些缺陷，尤其是当需要创建大量线程的时候：

线程生命周期的开销非常高 ：如果请求的到达率非常高，且处理过程是轻量级的，那么没创建和销毁一个新线程将消耗大量的计算资源。
资源消耗 ：活跃的线程会消耗系统资源，尤其是内存。如果可运行的线程数量大于可用处理器的数量，那么有些线程将会闲置，而且大量线程在竞争cpu资源还会产生其他的性能开销。
稳定性 ：在可创建线程的数量上存在一个限制。这个限制随着平台的不同的，受到多个制约因素，包括：
- JVM的启动参数。
- Thread构造函数中请求栈的大小
- 以及底层的操作系统对线程的限制等。如果破坏了这些限制，可能抛出OOM。

在一定范围内，增加线程可以提升系统的吞吐率，但是如果超过了这个范围，再创建线程只会降低系统的执行速度，并且如果过多地创建一个线程，那么整个应用也许都会崩溃。

Executor框架

任务是一组逻辑工作单元，而线程是使任务异步执行的机制。

线程池优化了线程管理工作，并且java.util.concurrent提供了一种灵活的线程池实现作为Executor框架的一部分。在java类库中，任务执行的主要抽象不是Thread，而是Executor.

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

它提供了一种标准的方法将任务的提交过程与执行过程解耦，并用Runnable表示任务。

Executor的实现还提供了对生命周期的支持，以及统计信息收集、应用程序管理机制和性能坚实等机制。

Executor基于生产者和消费者模式，提交任务的操作相当于生产者（生产待完成的工作单元），执行任务的操作相当于消费者。

基于Executor的web服务器

标准的Executor实现：

// 创建线程池
static final Executor executor = Executors.newFixedThreadPool(num);

// 创建任务 
Runnable task = new Runnable(){
    public void run(){
        doSomething();
    }
}

// 执行线程
executor.execute(task);

通过使用Executor，将请求任务的提交与任务的实际行为解耦，并只需要采用另一种不同的Executor实现，就可以改变服务器行为。

执行策略

通过将任务的提交和执行解耦开，就可以无需太大的困难为某种类型的任务指定或修改执行策略。最佳策略取决于可用的计算资源以及对服务质量的需求。通过限制并发任务的数量，可以确保应用程序不会由于资源耗尽而失败，或者由于在稀缺资源的竞争而影响性能。

每当看到 new Thread(Runnable).start() 时，并且你希望获得一个更加灵活的执行策略时，请使用Executor来代替Thread

线程池

线程池是指管理同一组同构工作线程的资源池。线程池是与工作队列密切相关的，工作队列中保存了所有等待执行的任务。工作者线程的任务很简单：从工作队列中获取一个任务，执行任务，然后回到线程池等待下一个任务。

通过重用现有线程而不是创建新线程，可以在处理多个请求的时候分摊掉创建和销毁线程的成本。而且在请求到达的时候，工作线程一般已经存在，就不需要等待线程创建的时间，提高了响应性。通过限制线程池大小，还可以避免多线程之间过度竞争资源，导致程序耗尽内存。

Executor的静态工厂方法

newFixedThreadPool
newCacheThreadPool
newSingleThreadExecutor
newScheduledThreadPool

newFixedThreadPool和newCacheThreadPool这两个方法返回通用的ThreadPoolExecutor实例，可以用来构建专门用途的executor。

Executor的生命周期

Executor的实现通常会创建线程来执行任务。但JVM只有在所有的非守护线程全部终止后才会退出。如果无法正确的关闭Executor，那么JVM将无法结束。

为了解决执行服务生命周期的问题，ExecutorService拓展了Executor接口，添加了一些生命周期的管理方法。

public interface ExecutorService extends Executor{
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout,TimeUnit unit) throws InterruptedException;
}

ExecutorService的生命周期有3种状态：运行，关闭和终止。

shutdown()
shutdownNow()

在ExecutorService关闭后提交的任务将由 拒绝执行处理器（Rejected Execution Handle） 来处理，它会抛弃任务，或使execute方法抛出一个未检查的RejectedExecutionException。等所有任务完成后，ExecutorService将转入终止状态。

延迟任务与周期任务

Timer 类负责管理延迟任务（“在100ms后执行该任务”）以及周期任务（“每100ms执行一次该任务”）。然而，Timer存在一些缺陷，因此应该考虑使用 ScheduledThreadPoolExecutor 来替代它（Timer支持的是绝对时间而不是相对时间的调度制度，因此任务的执行对系统时间非常敏感。而ScheduledThreadPoolExecutor只支持系统相对时间）。

Timer在执行所有定时任务时只会创建一个线程。如果某个任务的执行时间过长，那么会破坏其他timerTask的定时精准性。例如某个周期TimerTask需要每10ms执行一次，而另一个Task执行了50ms，那么TimerTask会在50ms以后快速的连续调用5次，或者直接丢掉这5次执行。（取决于Timer是基于固定速率还是说基于固定延时来调度）。

Timer的另一个问题是，如果TimerTask抛出了一个未检查的异常，Timer线程并不捕获异常。因此当TimerTask抛出未检查的异常时，将终止定时线程。这种情况下，Timer不会恢复线程的执行，而是会错误地认为整个Timer都被取消了。因此已经被调度但是未执行的Task不会被执行，新的任务也不会被调度。

现在基本不会使用Timer

找出可利用的并行性

Executor框架帮助指定执行策略，但是如果要使用Executor，必须将任务表述为一个Runnable。在大多数服务器应用程序中都存在一个明显的任务边界：单个客户请求。但是很多边界并非明显可见的，即使是服务器应用程序，在用户请求中仍存在可发掘的并行性，例如数据库服务器。

携带结果的任务Callable和Future

Executor框架使用Runnable作为其基本的任务表示形式。Runnable是一种有很大局限的抽象，虽然run能写入到日志文件或将结果放入某个共享的数据结构，但是它 不能返回一个值或抛出一个受检查的异常 。

对于某些存在延迟的计算， Callable 是一种更好的抽象：它认为主入口点（即call）将返回一个值，并可能抛出一个异常。在Executor中包含了一些辅助方法，能将其他类型的任务封装为一个Callable，例如Runnable和java.security.PrivilegedAction。

Runnable和Callable描述的都是抽象的计算任务。这些任务通常是有范围的，即都有一个明确的起始点，并且最终都会结束。

Executor执行的任务有4个生命周期阶段：

创建
提交
开始
完成

在Executor框架中，已提交但尚未开始的任务可以取消，但对于那些已经开始执行的任务，只有当它们能响应中断时，才能取消。取消一个已经完成的任务不会有任何的影响。

Future表示一个任务的生命周期，并提供了相应的方法来判断是否已经完成取消，以及获取任务的结果和取消任务。

Future规范中包含的 隐含含义 是：任务的生命周期只能前进，不能后退，当某个任务完成后，它将永远的停留在完成状态上。

get方法的行为取决于任务的状态（尚未开始，正在运行，已完成）。

如果任务已经完成，那么get会立即返回或者抛出一个Exception。
如果任务没有完成，那么get将阻塞并直到任务完成。
如果任务抛出了异常，那么get将该异常封装为ExecutionException并重新抛出。
如果任务被取消，那么get将抛出CancellationException。
如果get抛出了ExecutorException，那么可以通过getCause获取被封装的初始异常。

public interface Callable<V>{
    V call() throws Exception;
}

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

可以通过很多方法创建一个Future来描述任务。ExecutorService中的所有submit方法都可以返回一个Future，从而将一个Runnable和Callable提交给Executor，并得到一个Future用来获得任务的执行结果或取消任务。还可以显式的为某个指定的Runnable或者Callable实例化一个FutureTask。

Future和Callable例子：

Callable<List<A>> task = new Callable<List<A>>(){
    public List<A> call(){
        List<A> list  = new ArrayList();
        return list;
    }
}

Future<List<A>> future = executor.submit(task);

List<A> list = future.get();

get方法拥有 状态依赖 的内在特性，因而调用者不需要知道任务的状态，此外在任务提交和获得结果中包含的安全发布属性也确保了这个方法是线程安全的。

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

通过对异构任务进行并行化来获得重大的性能提升是很困难的。如果没有在相似的任务之间找出细粒度的并行性，那么这种方法带来的好处就回减少。只有当大量相互独立且同构的任务可以并发进行处理时，才能体现出将程序的工作负载分配到多个任务中带来的真正的性能提升。

CompletionService:Executor与BlockingQueue

CompletionService 将Executor和BlockingQueue的功能融合在一起。可以将Callable任务提交给他来执行，然后使用类似于队列操作的take和poll等方法来获得已完成的结果，这个结果将会封装为Future。

ExecutorCompletionService 实现了CompletionService，并将计算部分委托给一个Executor。

ExecutorCompletionService的实现

QueueingFuture
done()

private class QueueingFuture<V> extends FutureTask<V> {
    QueueingFuture(Callable<V> c) { super(c); }
    QueueingFuture(Runnable t , V r) { super(t,r); }

    protected void done() {
        completionQueue.add(this);
    }
}

为任务设置时限

如果某个任务无法在指定时间内完成，如果将不再需要它的结果，此时可以放弃这个任务。但可能只会在指定的时间内等待数据，如果超出了时间，那么只显示已经获得的数据。

要实现这个功能，可以由任务本身来管理它的限定时间，并且在超时以后中止执行或取消任务。

此时可再次使用Future，如果一个限时的get方法抛出了 TimeoutException ,那么可以提前中止它，避免消耗更多资源。

long endNanos = System.nanoTime() + TIME_BUDGET;
Future<A> f = exec.submit(task);
long timeLeft = endNanos - System.nanoTime();
try{
    A a = f.get(timeLeft,NANOSECONDS);// timeLeft如果<=0，就会中断
} catch(ExecutionException e){

} catch(TimeoutException){
    f.cancel(true);
}