Java多线程学习笔记(四) 久处不厌

在Java多线程学习笔记的(一) (二)(三)篇, 我们已经基本对多线程有基本的了解了, 但是Runnable接口仍然是不完美的，如果你希望获取线程的执行结果的话，那可能要绕下路了。为了弥补的缺点，Java引入了Callable接口，Callable相当于一个增强版的Runnable接口. 但是如果说我想取消任务呢，Java就引入了Future类。但是Future还是不够完美，那就再增强:CompletionService.

• 异步执行框架Executor
  
• 再增强: CompletionService
  
• Callable的补充: FutureTask概述
  
• 可重复执行的异步任务: AsyncTask
  
• 参考资料
  

##Future和Callable Callable作为Runnable的增强版，可以使用ThreadPoolExecutor的另外一个submit方法来提交任务, 该submit方法的声明如下:

public <T> Future<T> submit(Callable<T> task)

Callable方法只有一个方法:

V call() throws Exception;

call方法的返回值代表相应任务的处理结果，其类型V是通过Callable接口的类型参数来指定的; call方法代表的任务在其执行过程中可以抛出异常。而Runnable接口中的run方法既无返回值也不能抛出异常。Executors.callable(Runnable task , T result)能够将Runnable接口转换为Callable接口实例。

接下来我们来将目光转移到返回值Future接口上,  Future接口实例可被看作是提交给线程池执行的任务的句柄，凭借Future实例我们就可以获得线程执行的返回结果。Future.get()方法可以用来获取task参数所制定任务的处理结果。该方法的声明如下:

V get() throws InterruptedException, ExecutionException;

Future.get()方法被调用时，如果相应的任务尚未执行完毕，那么Future.get()会使当前线程暂停，直到相应的任务执行结束(包括正常结束和抛出异常而终止)。因此Future.get()方法是一个阻塞方法，该方法能够抛出InterruptedException说明它可以响应中断。另外假设相应任务的执行过程中抛出了一个任意的异常。调用这个异常(ExecutionException)的getCause方法可返回任务执行中所抛出的异常。因此客户端代码可以通过捕获Future.get()调用抛出的异常来知晓执行过程中抛出的异常。由于任务在执行未完毕的时候调用Future.get()方法来获取该任务的处理结果会导致等待上下文切换，因此如果需要获取执行结果的任务应当尽早的向线程池提交，并且尽可能晚地调用Future.get()方法来获取任务的处理结果，而线程池则正好利用这段时间来执行已提交的任务.

Future接口还支持任务的取消:

boolean cancel(boolean mayInterruptIfRunning);

如果任务已经执行完毕或者已经被取消过了，或者因为其他原因不能被取消，该方法将会返回false. 如果成功取消，那么任务将不会被执行，任务的执行状态分为两种，一种是还未执行，一种是已经处于执行中。对于执行中的任务, 参数mayInterruptIfRunning代表是否通过中断去终止线程的执行。在这个方法执行完毕之后，返回true的话，与之相关联的调用, isDone和isCancelled将总会返回true。Future.isDone()方法可以检测响应的任务是否执行完毕。任务执行完毕、执行过程中抛出异常以及取消都会导致该方法返回true。Future.get()会使其执行线程无限制等待，直到相应的任务执行结束。但是有的时候处理不当就可能导致无限制的等待，为了避免这种无限制等待情况的发生，此时我们可以使用get方法另一个版本:

V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

指定等待时间，如果等待时间之内还没完成，该方法就会抛出TimeoutException。注意该方法参数指定的超时时间仅仅用于控制等待相应任务的处理结果最多等待多长时间，而非限制任务本身的执行时间。所以推荐的做法是客户端通常需要在捕获TimeoutException之后，取消该任务的执行。

下面是一个Callable 和 Future的示例:

public class FutureDemo {
    public static void main(String[] args) {
        ExecutorService threadPool = Executors.newCachedThreadPool();
        Future<String> future = threadPool.submit(() -> {
            TimeUnit.SECONDS.sleep(1);
            int i = 1 / 0;
            return "hello world";
        });
        String result = null;
        try {
            result = future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            // 打印出来的就是除0异常
            System.out.println(e.getCause());
        }
        System.out.println(result); // 直到Future任务执行结束,线程返回执行结果, 这行才会被执行
        threadPool.shutdownNow();
    }
}

异步执行框架Executor

Runnable和Callable接口是对任务执行逻辑的抽象，我们在对应的方法编写对应的任务执行逻辑就可以了.而java.util.concurretn.Executor接口则是对任务的执行进行的抽象，任务的提交方只需要调用Executor的execute方法便可以被执行，而无需关心具体的执行细节。 比如在任务执行之前做些什么，任务是采用一个专门的线程来执行的，还是采用线程池来执行的，多个任务是以何种顺序来执行的。可见Executor能够实现任务的提交与任务执行的具体细节解耦(Decoupling)。和对任务处理逻辑的抽象类似，对任务执行的抽象也能带给我们信息隐藏和关注点分离的好处。

那这个解耦所带来的好处是什么呢?  一个显而易见的例子就是在一定程度上能够屏蔽同步执行和异步执行的差异。对于同一个任务(Runnable或Callable的实例)，如果我们把这个任务提交给ThreadPoolExecutor(它实现了Executor)来执行，那它就是异步执行; 如果把它交给如下所示的Executor来执行:

public class SynchronousExecutor implements Executor {
    @Override
    public void execute(Runnable command) {
    }
}

那么该任务就是同步执行。那么该任务无论是同步执行还是异步执行对于调用方来说就没有太大的差异。Executor 接口比较简单，功能也比较有限:

• 它不能返回任务的执行结果给提交任务方
• Executor 的实现类常常会维护一定量的工作者线程。当我们不再需要Executor实例的时候，往往需要将其内部维护的工作者线程停掉以释放相应的资源，但是Executor 接口没有定义相应的方法。现在的需要就是我们需要加强Executor接口，但又不能直接改动Executor，那这里的做法就是做子接口，也就是ExecutorService，ExecutorService继承了Executor，内部也定义了几个submit方法，能够返回任务的执行结果。ExecutorService接口还定义了shutdown方法和shutdownNow来关闭对应的服务释放资源。

再增强: CompletionService

尽管Future接口使得我们能够方便地获取异步任务的处理结果，但是如果我们需要一次性的提交一批任务的处理结果的话，那么仅使用Future接口写出来的代码将颇为繁琐。CompletionService接口就是为解决这个问题而生。我们来简单的看下CompletionService提供的方法: submit方法和ThreadPoolExecutor的一个submit方法相同:

Future<V> submit(Callable<V> task);

如果要获取批量提交异步任务的执行结果，那么我们可以使用CompletionService接口专门为此定义的方法:

Future<V> take() throws InterruptedException;

该方法和BlockingQueue的take()相似，它是一个阻塞方法，其返回值是批量提交异步任务中一个已执行结束任务的Future实例。我们可以通过Future实例来获取异步任务的执行结果。如果调用take方法时没有已经执行完毕的任务，那么调用take方法的线程会被阻塞，直到有异步任务执行结束，调用take方法的线程才会被恢复执行。如果你想获取异步任务的结果，但是如果没有执行完毕的异步任务，那么CompletionService同样提供了非阻塞式的获取异步任务处理结果的方法:

• Futurepoll();  // 如果没有已完成的任务, 则返回null
• Futurepoll(long timeout, TimeUnit unit) throws InterruptedException; //  在指定的时间内没有已完成的异步任务, 返回null。Java标准库提供的CompletionService的实现类是ExecutorCompletionService，联系上文提到的Executor，我们可以大致推断这个类是Executor和CompletionService的结合体。我们随意找出一个ExecutorCompletionService的构造函数来大致看一下:
• public ExecutorCompletionService(Executor executor,  BlockingQueue<Future> completionQueue)

executor 是任务的执行者，任务执行结束后将执行结果Future实例放入BlockingQueue中

Callable的补充: FutureTask概述

无论是Runnable实例还是Callable实例所表示的任务，只要我们将其提交给线程池执行，那么这个任务我就可以认为是异步任务。采用Runnable实例来表示异步任务的优点是任务既可以交给一个专门的工作者线程来执行,也可以提交给线程池来执行，缺点是我们无法直接获取任务的执行结果。使用Callable实例来表示异步任务，其优点是我们可以通过ThreadPoolExecutor的返回值来获取任务的执行结果，但无法直接交给一个专门的工作者线程来执行。因此，使用Callable来表示异步任务会使任务的执行方式大大受限。

而java.util.concurrent的FutureTask类则融合了Runnable和Callable接口的优点: FutureTask是RunnableFuture的实现类， Runnable则是Runable和Future的子接口。因此FutureTask既是Runable的实现类也是Future接口的实现。FutureTask提供了一个构造函数可以将Callable实例转换为Runnable实例，该构造器的声明如下:

public FutureTask(Callable<V> callable)

如此就弥补了Callable实例只能交由线程池执行不能交给指定的工作者线程执行的缺点，而FutureTask实例本身也代表了我们要执行的任务，我们可以通过FutureTask实例来获取以前Runnable实例无法获取的线程执行结果。像是Executor.execute(Runnable  task)这样只认Runnable实例来执行任务的情况下依然可以获取执行结果。下面是一个示例:

public class FutureTaskDemo {
    public static void main(String[] args) throws Exception{
        FutureTask<String> futureTask = new FutureTask(()->{
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("hello world");
            return "hello world";
        });
        new Thread(futureTask).start();
        while (futureTask.isDone()){
            System.out.println(futureTask.get());
        }
    }
}

FutureTask还支持以回调的方式处理任务的执行结果。当FutureTask实例所代表的任务执行完毕之后，Future.done就会被执行，FutureTask.done的执行线程与FutureTask的run的执行线程是同一个线程，在FutureTask是空的protected方法，我们可以在子类中重写该方法以实现后续的回调处理，由于是在任务执行完毕之后done方法被触发，所以在done方法中调用get方法获取任务执行结果不会被阻塞。但是，由于任务的执行结束即包括正常结束，也包括异常终止以及任务被取消而导致的终止，因此FutureTask.done方法中的代码可能需要在调用FutureTask.get()前调用FutureTask.isCancelled来判断任务是否被取消，以免FutureTask.get调用抛出CancellationException异常。FutureTask的简单示例:

public class FutureTaskDemo extends FutureTask<String>{
    public FutureTaskDemo(Callable<String> callable) {
        super(callable);
    }
    public FutureTaskDemo(Runnable runnable, String result) {
        super(runnable, result);
    }
    /**
     * 任务执行结束该方法会自动被调用
     * 我们通过get方法获取线程的执行结果,
     * 由于该方法在任务执行结束被调用,所以该方法调用get不会被阻塞
     */
    @Override
    protected void done() {
        try {
            String threadResult = get();
            System.out.println(threadResult);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) {
        FutureTaskDemo futureTaskDemo = new  FutureTaskDemo(()->{
            System.out.println("hello world");
            return "hello world";
        });
        new Thread(futureTaskDemo).start();
    }
}

FutureTask仍然有一些不足之处，它基本上是被设计用来表示一次性执行的任务，其内部维护了一个表示任务运行状态(包括未开始运行，已经运行结束等)的状态变量，FutureTask.run在执行run的时候会先判断相应任务的执行状态: 如果已经执行过，那么会直接返回，并不会抛出异常。因此FutureTask实例所代表的任务是无法被重复执行的。这意味着同一个FutureTask实例不能多次提交执行，尽管不会有异常的抛出。FutureTask.runAndReset能够打破这种限制，使得一个FutureTask实例所代表的任务被多次执行，但是不记录任务的执行结果。因此，如果对同一个对象所表示的任务需要多次被执行，并且我们需要对任务的执行结果进行处理，那么FutureTask就不适用了，不过我们可以仿照FutureTask设计一个满足我们所需要的，这也是我们看源码的意义，重在体会和学习背后的设计思维，由此我们引出AsyncTask。示例:

public class FutureTaskDemo extends FutureTask<String>{
    public FutureTaskDemo(Callable<String> callable) {
        super(callable);
    }
    public FutureTaskDemo(Runnable runnable, String result) {
        super(runnable, result);
    }
    @Override
    public void run() {
        // 启动任务用runAndReset启用
        super.runAndReset();
    }
    /**
     * 任务执行结束该方法会自动被调用
     * 我们通过get方法获取线程的执行结果,
     * 由于该方法在任务执行结束被调用,所以该方法调用get不会被阻塞
     */
    @Override
    protected void done() {
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTaskDemo futureTaskDemo = new FutureTaskDemo(()->{
            System.out.println("hello world");
            return "hello world";
        });
        new Thread(futureTaskDemo).start();
        new Thread(futureTaskDemo).start();
    }
}

可重复执行的异步任务: AsyncTask

public abstract class AsyncTask<V> implements Runnable, Callable<V> {
    protected final Executor executor;
    public AsyncTask(Executor executor) {
        this.executor = executor;
    }
    public AsyncTask() {
        this(new Executor() {
            @Override
            public void execute(Runnable command) {
                command.run();
            }
        });
    }
    @Override
    public void run() {
        Exception exp = null;
        V r = null;
        try {
            //call的逻辑可以从传递也可以由自己来实现
           r = call();
        }catch (Exception e){
            exp = e;
        }
        final V result  = r;
        if (null == exp){
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    onResult(result);
                }
            });
        }else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    onError(result);
                }
            });
        }
    }
    /**
     *留给子类实现任务执行结果的处理逻辑
     * @param result
     */
    protected abstract void onError(V result);
    /**
     * 留给子类实现任务执行结果的处理逻辑
     * @param result
     */
    protected abstract void onResult(V result);
}