Java 异步任务执行服务（一）：基本概念和原理

1. 异步任务执行服务是什么意思

• 线程 Thread 既表示要执行的任务（run() 方法），又表示执行的机制（start() 方法）
• Java 并发包提供了一套框架，大大简化了执行异步任务所需的开发，这套框架引入了一个“执行服务”的概念
• 执行服务将任务的提交和任务的执行相分离。“执行服务”封装了任务执行的细节，对于任务提交者而言，它可以关注于任务本身，如提交任务、获取结果、取消任务；而不需要关注任务执行的细节，如线程的创建、任务调度、线程关闭等
• 任务执行服务体现了并发异步开发中关注点分离的思想，使用者只需要通过 ExecutorService 提交任务，通过 Future 操作任务和结果即可，不需要关注线程创建和协调的细节

2. 任务执行服务涉及的基本接口有哪些

• Runnable 和 Callable：表示要执行的异步任务
• Executor 和 ExecutorService：表示执行服务
• Future：表示异步任务的结果

3. Runnable 和 Callable 的区别是什么

• 二者都表示任务
  • Runnable 没有返回结果，而 Callable 有返回结果
  • Runnable 不会抛出异常，而 Callable 会抛出异常

4. 怎样理解 Executor

• Executor 表示最简单的执行服务
• 定义：public interface Executor { void execute(Runnable command); }，就是可以执行一个 Runnable，没有返回结果
• 没有限定任务如何执行，可能是创建一个新线程，可能是复用线程池中的某个线程，也可能是在调用者线程中执行

5. 怎样理解 ExecutorService

• ExecutorService 扩展了 Executor，定义了更多服务，基本方法有

  	public interface ExecutorService extends Executor {
  	    <T> Future<T> submit(Callable<T> task);
  	    <T> Future<T> submit(Runnable task, T result);
  	    Future<?> submit(Runnable task);
  	
  	    void shutdown();
  	    List<Runnable> shutdownNow();
  	    boolean isShutdown();
  	    boolean isTerminated();
  	    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
  	    <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;
  	    <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;
  	    <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
  }

• 三个 submit() 都表示提交一个任务，返回值类型都是 Future。返回后，只是表示任务已提交，不代表已执行，通过 Future 可以查询异步任务的状态、获取最终结果、取消任务等

  • 对于 Callable，任务最终有个返回值，对于 Runnable 是没有返回值的
  • 第二个提交 Runnable 的方法可以同时提供一个结果，在异步任务结束时返回
  • 第三个方法异步任务的最终返回值为 null

• 有两个关闭方法：shutdown() 和 shutdownNow()

  • 区别是，shutdown() 表示不再接受新任务，但已提交的任务会继续执行，即使任务还未开始执行
  • shutdownNow() 不仅不接受新任务，而且会终止已提交但尚未执行的任务，对于正在执行的任务，一般会调用线程的 interrupt() 方法尝试中断。不过，线程可能不响应中断，shutdownNow() 会返回已提交但尚未执行的任务列表
  • shutdown() 和 shutdownNow() 不会阻塞等待，它们返回后不代表所有任务都已结束。不过 isShutdown() 方法会返回 true。调用者可以通过 awaitTermination() 等待所有任务结束，它可以限定等待的时间，如果超时前所有任务都结束了，即 isTerminated() 方法返回 true，则返回 true，否则返回 false

• ExecutorService 有两组批量提交任务的方法：invokeAll() 和 invokeAny()，它们都有两个版本，其中一个限定等待时间

  • invokeAll() 方法等待所有任务完成，返回的 Future 列表中，每个 Future 的 isDone() 方法都返回 true。不过 isDone() 为 true 不代表任务就执行成功了，可能是被取消了。invokeAll() 可以指定等待时间，如果超时后有的任务没完成，就会被取消
  • 对于 invokeAny()，只要有一个任务在限时内成功返回了，它就会返回该任务的结果，其他任务会被取消；如果没有任务能在限时内成功返回，抛出 TimeoutException 异常；如果限时内所有任务都结束了，但都发生了异常，抛出 ExecutionException 异常

• 使用 ExecutorService，编写并发异步任务的代码就像写顺序程序一样，不用关心线程的创建和协调，只需要提交任务、处理结果就可以了，大大简化了开发工作

6. 怎样理解 Future

• 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;
  }

• get() 方法用于返回异步任务最终的结果，如果任务还未执行完成，会阻塞等待；带参数的 get() 方法可以限定阻塞等待的时间，如果超时任务还未结束，会抛出 TimeoutException 异常

• cancel() 用于取消异步任务，如果任务已完成、或已经取消、或由于某种原因不能取消，cancel() 返回 false，否则返回 true。如果任务还未开始，则不再运行。但如果任务已经在运行，则不一定能取消，参数 mayInterruptIfRunning 表示，如果任务正在执行，是否调用 interrupt() 方法中断线程，如果为 false，就不会，如果为 true，就会尝试中断线程，虽然中断不一定能取消线程

• isDone() 和 isCancelled() 用于查询任务状态。isCancelled() 表示任务是否被取消，只要 cancel() 方法返回了 true，随后的 isCancelled() 方法都会返回 true，即使执行任务的线程还未真正结束。isDone() 表示任务是否结束，不管什么原因都算，可能是任务正常结束，可能是任务抛出了异常，也可能是任务被取消

• 对于 get() 方法，如果调用get() 方法的线程被中断了，get() 方法会抛出 InterruptedException 异常，任务最终大概有三种结果

  • 正常完成，get() 方法会返回其执行结果，如果任务是 Runnable 且没有提供结果，返回 null
  • 任务执行抛出了异常，get() 方法会将异常包装为 ExecutionException 重新抛出，通过异常的 getCause() 方法可以获取原异常
  • 任务被取消了，get() 方法会抛出异常 CancellationException

• Future 是一个重要的概念，是实现“任务的提交”与“任务的执行”相分离的关键，是其中的“纽带”，任务提交者和任务执行服务通过它隔离各自的关注点，同时进行协作

7. 写一个任务执行服务 Demo

public class BasicDemo {
    static class Task implements Callable<Integer> {
        @Override
        public Integer call() throws Exception {
            int sleepSeconds = new Random().nextInt(1000);
            Thread.sleep(sleepSeconds);
            return sleepSeconds;
        }
    }
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newSingleThreadExecutor(); //使用工厂类 Executors 创建了一个任务执行服务，表示使用一个线程执行所有服务
        Future<Integer> future = executor.submit(new Task());
        //模拟执行其他任务
        Thread.sleep(100);
        try {
            System.out.println(future.get());
        } catch(ExecutionException e) {
            e.printStackTrace();
        }
        executor.shutdown(); //关闭任务执行服务
    }
}

8. ExecutorService 的基本实现原理

• ExecutorService 的主要实现类是 ThreadPoolExecutor，它是基于线程池实现的
• ExecutorService 还有一个抽象实现类 AbstractExecutorService
• AbstractExecutorService 提供了 submit()、invokeAll() 和 invokeAny() 的默认实现，子类需要实现其他方法
  • 除了 execute()，其他方法都与执行服务的生命周期管理有关
  • submit()、invokeAll()和 invokeAny() 最终都会调用 execute()，execute() 决定了到底如何执行任务

9. Future 的基本实现原理

• Future 的主要实现类是 FutureTask
• FutureTask 实现了 RunnableFuture 接口
• RunnableFuture 接口既扩展了 Runnable，又扩展了 Future，没有定义新方法
  • 作为 Runnable，它表示要执行的任务，传递给 execute() 方法进行执行
  • 作为 Future，它又表示任务执行的异步结果

10. 写一个简单的 ExecutorService 实现类

public class SimpleExecutorService extends AbstractExecutorService {
    @Override
    public void shutdown() {
    }
    
    @Override
    public List<Runnable> shutdownNow() {
        return null;
    }
    
    @Override
    public boolean isShutdown() {
        return false;
    }
    
    @Override
    public boolean isTerminated() {
        return false;
    }
    
    @Override
    public boolean awaitTermination(long itmeout, TimeUnit unit) throws InterruptedException {
        return false;
    }
    
    @Override
    public void execute(Runnable command) {
        new Thread(command).start();
    }
}