Executor框架 - HelloWorld开发者社区

任务是一组逻辑工作单元，而线程则是使任务异步执行的机制。线程池简化了线程的管理工作，并且java.util.concurrent提供了一种灵活的线程池实现作为Executor框架的一部分。在Java类库中，任务执行的主要抽象不是Thread，而是Executor，如下所示：

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

虽然Executor是个简单的接口，但它却为灵活且强大的异步任务执行框架提供了基础，该框架能支持多种不同类型的任务执行策略。它提供了一种标准的方法将任务的提交过程与执行过程解耦开来，并用Runnable来表示任务。Executor的实现还提供了对生命周期的支持，以及统计信息收集、应用程序管理机制和性能监视等机制。

Executor基于生产者--消费者模式，提交任务的操作相当于生产者（生成待完成的工作单元），执行任务的线程则相当于消费者（执行完这些工作单元）。如果要在程序中实现一个生产者--消费者模式的设计，那么最简单的方式通常就是使用Executor。

示例：基于Executor的Web服务器

class TaskExecutionWebServer {
    private static final int NTHREADS = 100;
    private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);
    
    public static void main(String[] args) throws IOException {
        ServerSocket socket = new ServerSocket(80);
        while(true) {
            final Socket connection = socket.accept();
            Runnable task = new Runnable() {
                public void run() {
                    handleRequest(connection);
                }
            );
            exec.execute(task);
        }
    }
}

在TaskExecutionWebServer中，通过使用Executor，将请求处理任务的提交与任务的实际执行解耦开来，并且只需采用另一种不同的Executor实现，就可以改变服务器的行为。改变Executor的实现或配置所带来的影响要远远小于改变任务提交方式带来的影响。通常，Executor的配置是一次性的，因此在部署阶段可以完成，而提交任务的代码却会不断地扩散到整个程序中，增加了修改的难度。

执行策略

通过将任务的提交与执行解耦，从而无需太大的困难就可以为某种类型的任务指定和修改执行策略。在执行策略中定义了任务执行的“what, where, when, how”等方面，包括：

在什么（What）线程中执行任务？
任务按照什么（What）顺序执行（FIFO, LIFO, 优先级）？
有多少个（How many）任务能并发执行？
在队列中有多少个（How many）任务在等待执行？
如果系统由于过载而需要拒绝一个任务，那么应该选择哪一个（Which）任务？另外，如何（How）通知应用程序有任务被拒绝？
在执行一个任务之前或之后，应该进行哪些（What）操作？

各种执行策略都是一种资源管理工具，最佳策略取决于可用的计算资源以及对服务质量的需求。通过限制并发任务的数量，可以确保应用程序不会由于资源耗尽而失败，或者由于在稀缺资源上发生竞争而严重影响性能。通过将任务的提交与执行策略分离开来，有助于在部署阶段选择与可用硬件资源最匹配的策略执行。

每当看到下面这种形式的代码时：
new Thread(runnable).start();
并且你希望获得一种更灵活的执行策略时，请考虑使用Executor来代替Thread。

线程池

线程池，从字面含义来看，是指管理一组同构工作线程的资源池。线程池是与工作队列（Work Queue）密切相关的，其中在工作队列重保存了所有等待执行的任务。工作者线程（Worker Thread）的任务很简单：从工作队列中获取一个任务，执行任务，然后返回线程池并等待下一个任务。

“在线程池重执行任务”比“为每一个任务分配一个线程”优势更多。通过重用现有的线程而不是创建新线程，可以在处理多个请求是分摊在线程创建和销毁过程中产生的巨大开销。另一个额外的好处是，当请求到达时，工作线程通常已经存在，因此不会由于等待创建线程而延迟任务的执行，从而提高了响应性。通过适当调整线程池的大小，可以创建足够多的线程以便使处理器保持忙碌状态，同时还可以防止过多线程相互竞争资源而使应用程序耗尽内存或失败。

类库提供了一个灵活的线程池以及一些有用的默认配置。可以通过调用Executors中的静态工厂方法之一来创建一个线程池：

newFixedThreadPool：将创建一个固定长度的线程池，每当提交一个任务时就创建一个线程，直到达到线程池的最大数量，这时线程池的规模将不再变化（如果某个线程由于发生了未预期的Exception而结束，那么线程池会补充一个新的线程）。
newCachedThreadPool：将创建一个可缓存线程的线程池。根据任务按需创建线程，并且当任务结束后会将该线程缓存60秒，如果期间有新的任务到来，则会重用这些线程，如果没有新任务，则这些线程会被终止并移除缓存。此线程池适用于处理量大且短时的异步任务。
newSingleThreadExecutor：是一个单线程的Executor，它创建单个工作者线程来执行任务，如果这个线程异常结束，会创建另一个线程来替代。它还能确保依照任务在队列中的顺序来串行执行。
newScheduledThreadPool：创建了一个固定长度的线程池，而且以延迟或定时的方式来执行任务，类似于Timer。

从“为每一个任务分配一个线程”策略变成基于线程池的策略，将对应用程序的稳定性产生重大的影响：Web服务器不会再在高负载情况下失败。由于服务器不会创建数千个线程来争夺有限的CPU和内存资源，因此服务器的性能将平缓的降低。通过使用Executor，可以实现各种调优、管理、监视、记录日志、错误报告和其他功能。

Executor的生命周期

我们已经知道如何创建一个Executor，但并没有讨论如何关闭它。Executor的实现通常会创建线程来执行任务。但JVM只有在所有（非守护）线程全部终止后才会退出。因此，如果无法正确的关闭Executor，那么JVM将无法结束。

由于Executor以异步方式来执行任务，因此在任何时刻，之前提交任务的状态不是立即可见的。有些任务可能已经完成，有些可能正在运行，而其他的任务可能在队列中等待执行。当关闭应用程序时，可能采用最平缓的关闭形式（完成所有已经启动的任务，并且不再接受任何新的任务），也可能采用最粗暴的关闭形式（直接关掉机房的电源），以及其他各种可能的形式。既然Executor是为应用程序提供服务的，因而它们也是可关闭的（无论采用平缓的方式还是粗暴的方式），并将关闭操作中受影响的任务的状态反馈给应用程序。

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

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

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

在ExecutorService关闭后提交的任务将由“拒绝执行处理器（Rejected Execution Handler）”来处理，它会抛弃任务，或者使得execute方法抛出一个未检查的RejectedExecutionException。等所有任务都完成后，ExecutorService将进入终止状态。可以调用awaitTermination来等待ExecutorService到达终止状态，或者通过调用isTerminated来轮询ExecutorService是否已经终止。通常在调用awaitTermination之后会立即调用shutdown，从而同步关闭ExecutorService。

LifecycleWebServer通过增加生命周期来支持扩展Web服务器的功能。可以通过两种方法来关闭Web服务器：在程序中调用stop方法，或者以客户端请求的形式向Web服务器发送一个特定格式的HTTP请求：

class LifecycleWebServer {
    private final ExecutorService exec = ...
    public void start() throws IOException {
        ServerSocket socket = new ServerSocket(80);
        while(!exec.isShutdown()) {
            try {
                final Socket connection = socket.accept();
                exec.execute(new Runnable() {
                    public void run() {
                        handleRequest(connection);
                    }
                ));
            } catch(RejectedExecutionException e) {
                if (!exec.isShutdown) {
                    log("Task submission rejected", e);
                }
            }
        }
    }
    
    //通过程序中调用stop方式来停止WebServer
    public void stop() {
        exec.shutdown();
    }
    
    //通过向WebServer发送特定的HTTP请求来停止WebServer
    void handleRequest(Socket connection) {
        Request rq = readRequest(connection);
        if (isShutdownRequest) {
            stop();
        } else {
            dispatchRequest(rq);
        }
    }
}

延迟任务与周期任务

Timer类负责管理延迟任务（“在10ms后执行该任务”）以及周期任务（“每10ms执行一次该任务”）。然而，Timer存在一些缺陷，因此应该考虑使用ScheduledThreadPoolExecutor来代替它（Timer支持基于绝对时间而不是相对时间的调度机制，因此任务的执行对系统时钟变化很敏感，而ScheduledThreadPoolExecutor只支持基于相对时间的调度）。可以通过ScheduledThreadPoolExecutor的构造函数或newScheduledThreadPool工厂方法来创建该类的对象。

Timer在执行所有定时任务时只会创建一个线程。如果某个任务的执行时间过长，那么将破坏其他TimerTask的精确性。例如某个周期TimerTask需要每10ms执行一次，而另一个TimerTask需要执行40ms，那么这个周期任务要么在40ms任务执行后快速连续地调用4次，要么将彻底“丢失”4次调用（取决于它是基于固定速率来调度还是基于固定延迟来调度）。线程池能弥补这个缺陷，它可以提供多个线程来执行延迟任务和周期任务。

Timer的另一个问题是，如果TimerTask抛出了一个未检查的异常，那么Timer将表现出糟糕的行为。Timer线程并不捕捉异常，因此当TimerTask抛出异常时，将终止定时线程。这种情况下，Timer也不会恢复线程的执行，而是会错误的认为整个Timer都被取消了。因此已经被调度单稍微执行的TimerTask将不会再执行，新的任务也不能被调度（这个问题称为“线程泄露[Thread Leake]”）。

程序OutOfTime中给出了Timer中为什么会出现这种问题，以及如何使得试图提交TimerTask的调用者也出现问题。你可能认为程序会运行6秒后退出，但实际情况是运行1秒就结束了，并抛出一个异常信息“Timer already cancelled”。 ScheduledThreadPoolExecutor能正确处理这些表现出错误行为的任务。在Java 5.0或更高版本的JDK中，将很少使用Timer。

public class OutOfTime {
    public static void main(String[] args) throws Exception {
        Timer timer = new Timer();
        //1ms后执行ThrowWoker，会抛出RuntimeException导致Timer异常结束
        timer.schedule(new ThrowWoker(), 1);
        SECONDS.sleep(1);

        //当Timer已经退出后再次调用timer的schedule方法时，则会报错
        timer.schedule(new ThrowWoker(), 1);
        SECONDS.sleep(5);
    }
    
    static class ThrowWoker extends TimerTask {
        public void run() {
            throw new RuntimeException();
        }
    }
}

如果要构建自己的调度任务，那么可以使用DelayQueue，它实现了BlockingQueue，并未ScheduledThreadPoolExecutor提供调度功能。DelayQueue管理着一组Delayed对象，每个对象都有一个相应的延迟时间：在DelayQueue中，只有某个元素逾期后，才能从DelayQueue执行take操作。从DelayQueue中返回的对象将根据它们的延迟时间进行排序。