ExecutorService

继承树

ExecutorService是java对线程池定义的一个接口，也是java默认给我们的，它具有两个实现类：

ThreadPoolExecutor
ScheduledThreadPoolExecutor

ExecutorService继承了Executor接口，这个接口只有一个excute()方法，这里我们展示一下它们的类图：

在这里插入图片描述

原理

一般在实际业务中，我们对于线程的使用都是比较频繁的，但是线程之间的切换是非常消耗cpu的，所以我们会使用线程池，也就是java提前创建好了许多线程，利用可回收的池化的概念，来完成对线程池的优化的工作。

创建ExecutorService

java给我们提供了Executors工厂类，可以帮助我们创建各种类型的ExecutorService线程池，如下四种：

1. newCachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
2. newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
3. newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
4. newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

这里值得注意的是，Executors只是一个工厂类，它所有的方法返回的都是ThreadPoolExecutor、ScheduledThreadPoolExecutor这两个类的实例。

使用ExecutorService

ExecutorService executorService = Executors.newFixedThreadPool(10);

executorService.execute(new Runnable() {
public void run() {
    System.out.println("Asynchronous task");
}
});

executorService.shutdown();

执行ExecutorService

ExecutorService的执行方法有四种

- execute(Runnable)
- submit(Runnable)
- submit(Callable)
- invokeAny(...)
- invokeAll(...)

execute(Runnable)

这个方法可以接受一个runnable实例，并且异步执行，但是这个方法没有办法知道task的执行结果：

ExecutorService executorService = Executors.newSingleThreadExecutor();

executorService.execute(new Runnable() {
public void run() {
    System.out.println("Asynchronous task");
}
});

executorService.shutdown();

submit(Runnable)

submit(Runnable)也是接收一个runnable实例，不同的是，这个方法可以返回一个Future对象，通过返回的Future对象，我们可以检查提交的任务是否执行完成：

ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future future = executorService.submit(new Runnable() {
            public void run() {
                System.out.println("Asynchronous task");
            }
        });

        System.out.println(future.get());//如果这里返回一个null，表示任务执行完毕，这个方法会产生阻塞。

submit(Callable)

这个方法接收的是Callable实例，然后也会返回一个Future对象。除此之外，因为这个方法接收的是Callable接口的一个实例，这个接口会有一个call（）方法的返回值，可以返回任务的执行结果。

 ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future future = executorService.submit(new Callable(){
            public Object call() throws Exception {
                System.out.println("Asynchronous Callable");
                return "Callable Result";
            }
        });

        System.out.println("future.get() = " + future.get());

如果任务执行完成，future.get()方法会返回Callable任务的执行结果。注意，future.get()方法会产生阻塞。

invokeAny（Callable）

invokeAny（Callable）接收一个Callable集合，执行的方法不会返回一个Future，但是会返回所有的callable任务中其中一个任务的执行结果，但是不能保证返回的是哪一个执行的结果，反正就是其中一个。

ExecutorService executorService = Executors.newSingleThreadExecutor();

Set<Callable<String>> callables = new HashSet<Callable<String>>();

callables.add(new Callable<String>() {
public String call() throws Exception {
    return "Task 1";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
    return "Task 2";
}
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
    return "Task 3";
}
});

String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();

invokeAll（Callable）

invokeAll（Callable）也是接收一个Callable集合，然后返回一个Future的list，对应每一个任务执行后的Future对象

ExecutorService executorService = Executors.newSingleThreadExecutor();

Set<Callable<String>> callables = new HashSet<Callable<String>>();

callables.add(new Callable<String>() {
public String call() throws Exception {
    return "Task 1";
}
});
callables.add(new Callable<String>() {
    public String call() throws Exception {
    return "Task 2";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
    return "Task 3";
}
});

List<Future<String>> futures = executorService.invokeAll(callables);

for(Future<String> future : futures){
System.out.println("future.get = " + future.get());
}

executorService.shutdown();

关闭ExecutorService

当我们使用完成ExecutorService之后应该关闭它，否则它里面的线程会一直处于运行状态。
举个例子，如果的应用程序是通过main()方法启动的，在这个main()退出之后，如果应用程序中的ExecutorService没有关闭，这个应用将一直运行。之所以会出现这种情况，是因为ExecutorService中运行的线程会阻止JVM关闭。
如果要关闭ExecutorService中执行的线程，我们可以调用ExecutorService.shutdown()方法。在调用shutdown()方法之后，ExecutorService不会立即关闭，但是它不再接收新的任务，直到当前所有线程执行完成才会关闭，所有在shutdown()执行之前提交的任务都会被执行。
如果我们想立即关闭ExecutorService，我们可以调用ExecutorService.shutdownNow()方法。这个动作将跳过所有正在执行的任务和被提交还没有执行的任务。但是它并不对正在执行的任务做任何保证，有可能它们都会停止，也有可能执行完成。

使用RunTime.getRunTime().addShutdownHook优雅关闭线程池

有时候我们用到的程序不一定总是在JVM里面驻守,可能调用完就不用了,释放资源.
RunTime.getRunTime().addShutdownHook的作用就是在JVM销毁前执行的一个线程.当然这个线程依然要自己写.
利用这个性质,如果我们之前定义了一系列的线程池供程序本身使用,那么就可以在这个最后执行的线程中把这些线程池优雅的关闭掉.

比如我们定义了一个线程池

private ExecutorService streamThreadPool = Executors.newFixedThreadPool(streamNum);

我们对它优雅的关闭

Runtime.getRuntime().addShutdownHook(new Thread() {
   public void run() {
      shutdownGracefully();
   }
});

public void shutdownGracefully() {
   shutdownThreadPool(streamThreadPool, "main-pool");
}

/**
 * 优雅关闭线程池
 * @param threadPool
 * @param alias
 */
private void shutdownThreadPool(ExecutorService threadPool, String alias) {
   log.info("Start to shutdown the thead pool: {}", alias);

   threadPool.shutdown(); // 使新任务无法提交.
   try {
      // 等待未完成任务结束
      if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
         threadPool.shutdownNow(); // 取消当前执行的任务
         log.warn("Interrupt the worker, which may cause some task inconsistent. Please check the biz logs.");

         // 等待任务取消的响应
         if (!threadPool.awaitTermination(60, TimeUnit.SECONDS))
            log.error("Thread pool can't be shutdown even with interrupting worker threads, which may cause some task inconsistent. Please check the biz logs.");
      }
   } catch (InterruptedException ie) {
      // 重新取消当前线程进行中断
      threadPool.shutdownNow();
      log.error("The current server thread is interrupted when it is trying to stop the worker threads. This may leave an inconcistent state. Please check the biz logs.");

      // 保留中断状态
      Thread.currentThread().interrupt();
   }

   log.info("Finally shutdown the thead pool: {}", alias);
}