【异步】Futurn、FutureTask、CompletionService、CompletableFuture

偏执的太偏执、 2023-10-12 12:49 43阅读 0赞

### 1. Callable ###

在这篇文章中 [【Thread】线程的基本概念及创建方式（一）][Thread]，我们知道创建线程的几种方式。其中，有两个是通过接口来实现的：`Runnable`、`Callable`。它们的区别如下：

*  `Runnable` 接口中的方法是没有返回值的，而 `Callable` 接口中的方法是有返回值的
 *  `Callable` 接口中的方法是可以抛异常的

**如何使用 `Callable` 接口**

有两种方式可以使用 `Callable` 接口：

1.  通过 `Thread`，再结合 `Future` 、`FutureTask`
2.  通过线程池的 `submit()`、`invokeAll()`、`invokeAny()` 方法

方式一：

public static void main(String[] args) throws Exception {
        
        Callable<String> task = () -> "I am a callable Task";
        // 将 Callable 封装为 FutureTask
        FutureTask<String> futureTask = new FutureTask<>(task);
        new Thread(futureTask).start();
        String result = futureTask.get();
    }

注意：`get()` 方法是一个阻塞方法，没有获取到结果之前，一直处于阻塞状态！

方式二：

public static void main(String[] args) throws Exception {
        
        Callable<String> task = () -> "I am a callable Task";
        ExecutorService threadPool = Executors.newFixedThreadPool(2);
        Future<String> future = threadPool.submit(task);
        String result = future.get();
        threadPool.shutdown();
    }

### 2. Future 、FutureTask ###

首先，看看它们之间的关系图，如下：

![在这里插入图片描述][e89347c3d79d40d397ddb5fb9eeb2f84.png]

**`Future` 接口**： Future 就是对于具体的 Runnable 或者 Callable 任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过 `get()` 方法获取执行结果，该方法会阻塞直到任务返回结果

public interface Future<V> {
        
        // 取消任务，如果任务正在运行的，mayInterruptIfRunning为true时，表明这个任务会被打断的，并返回true；
        // 为false时，会等待这个任务执行完，返回true；若任务还没执行，取消任务后返回true，如任务执行完，返回false
        boolean cancel(boolean mayInterruptIfRunning);
        // 判断任务是否被取消了,正常执行完不算被取消
        boolean isCancelled();
        // 判断任务是否已经执行完成，任务取消或发生异常也算是完成，返回true
        boolean isDone();
        // 获取任务返回结果，如果任务没有执行完成则等待完成将结果返回，如果获取的过程中发生异常就抛出异常，
        // 比如中断就会抛出InterruptedException异常等异常
        V get() throws InterruptedException, ExecutionException;
        // 在规定的时间如果没有返回结果就会抛出TimeoutException异常
        V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
    }

**`FutureTask` 接口**：`FutureTask` 实现了 `RunnableFuture` 接口，而此接口又继承了 `Runnable` 接口、`Future` 接口。所以，必然会重新 `run()` 方法、`Future` 接口中的方法

public class FutureTask<V> implements RunnableFuture<V> {
        
    
    	private Callable<V> callable;
    	// 存储任务结果
    	private Object outcome;
    	
    	public void run() {
        
            if (state != NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
                return;
            try {
        
                Callable<V> c = callable;
                if (c != null && state == NEW) {
        
                    V result = c.call();
                    set(result);
                }
            } finally {
        
                //...
            }
        }
        
        protected void set(V v) {
        
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        
            	// 将结果赋值给 outcome
                outcome = v;
                // 修改线程状态
                UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
                // 移除所有等待线程并发出信号
                finishCompletion();
            }
        }
        
        private void finishCompletion() {
        
            for (WaitNode q; (q = waiters) != null;) {
        
                if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
        
                    for (;;) {
        
                        Thread t = q.thread;
                        if (t != null) {
        
                            q.thread = null;
                            LockSupport.unpark(t);
                        }
                        WaitNode next = q.next;
                        if (next == null)
                            break;
                        q.next = null; // unlink to help gc
                        q = next;
                    }
                    break;
                }
            }
    		// 留给子类重写
            done();
            callable = null;        // to reduce footprint
        }
    }

`FutureTask.run()` 方法的逻辑是：通过调用 `Callable.call()` 方法去执行任务，并将返回的结果通过 `set()` 方法赋值给 `outcome` 属性。

那么，上述的代码块：`new Thread(futureTask).start();`的流程是：

> 通过 Thread 的 start() 方法执行了 FutureTask 的 run() 方法，最终通过 Callable 的 call() 方法执行了！

`get()` 方法的实现： `get()` 方法里会先通过 `state` 变量判断任务是否已跑完：若跑完，则直接将结果返回；否则，就构造等待节点，并扔进等待队列**自旋**，阻塞主线程。另一边的线程计算出结果后就会将等待队列里的所有节点依次出队并唤醒线程

public V get() throws InterruptedException, ExecutionException {
        
        int s = state;
        // 任务还未执行完，等待执行完
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        // 任务执行完|已取消，返回结果或者抛出异常
        return report(s);
    }
    
    private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
        
        	// 发生线程中断，从等待队列中移除节点，并抛出中断异常
            if (Thread.interrupted()) {
        
                removeWaiter(q);
                throw new InterruptedException();
            }
    
            int s = state;
            // 任务已执行完|已取消
            if (s > COMPLETING) {
        
            	// 若已有等待节点，将线程置空，避免另外一边再次调用 unpark()
                if (q != null)
                    q.thread = null;
                return s;
            }
            // 任务正在执行，让出时间片
            else if (s == COMPLETING)
                Thread.yield();
            // 如果还未构造等待队列节点，则构造
            else if (q == null)
                q = new WaitNode();
            // 如果为入队，则 CAS 尝试入队；如果入队失败，则下一次循环再次尝试
            else if (!queued)
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            // 通过 LockSupport.park() 阻塞线程
            else if (timed) {
        
                nanos = deadline - System.nanoTime();
                // 如果限时，且不大于 0，则从等待队列中移除节点，并返回
                if (nanos <= 0L) {
        
                    removeWaiter(q);
                    return state;
                }
                LockSupport.parkNanos(this, nanos);
            }
            else
                LockSupport.park(this);
        }
    }
    
    private V report(int s) throws ExecutionException {
        
        Object x = outcome;
        if (s == NORMAL)
            return (V)x;
        //...
    }

`get()` 方法处理逻辑如下：

1.  先判断 state 的值是否大于 comleting，也就是执行完，是的话就判断 state 为 normal 正常的话，就返回 outcome 结果，为 cancel 取消的话，就抛取消异常，为异常的话，就抛出 outcome 记录的异常
2.  如果 state 的值表示还没执行完的话，就会进入一个自旋操作，也就是一个死循环
3.  如果 state 是 completing 的话，就通过 Thread.yield 让出时间片
4.  否则如果结果已经出来了，那就返回
5.  否则如果还未构造等待结点，那就构造等待结点，结点里保存了当前线程
6.  否则如果结点还未入队，那就将结点入队
7.  否则调用 `LockSupport.park()` 方法阻塞线程

执行任务的线程那边在执行完任务后(执行完 `run()` 方法)，就把结果或者异常保存到 outcome 里，修改 state 的值，并且依次将队列里的结点出队，并调用 `LockSupport.unpark()` 方法唤醒节点里的线程（这些操作是在 `set()` 方法中）

> 案例：使用 Future 来批量执行多个异步任务的业务并获取结果

ExecutorService executor = Executors.newFixedThreadPool(2); 
    Future f1 = excutor.submit(c1);
    f1.get();
    Future f2 = excutor.submit(c2);
    f2.get();

`f1.get()` 在获取成功之前会被阻塞，会阻塞 c2 的执行，严重降低了效率

`Future` 接口局限性：

从本质上说，Future 表示一个异步计算的结果。它提供了 isDone() 来检测计算是否已经完成，并且在计算结束后，可以通过get() 方法来获取计算结果。但它本身也确实存在着许多限制：

1.  **并发执行多任务**：Future 只提供了 get() 方法来获取结果，并且是阻塞的
2.  **无法对多个任务进行链式调用**：如果你希望在计算任务完成后执行特定动作，比如发邮件，但 Future 却没有提供这样的能力
3.  **无法组合多个任务**：如果你运行了10个任务，并期望在它们全部执行结束后执行特定动作，那么在 Future 中这是无能为力的
4.  **没有异常处理**：Future 接口中没有关于异常处理的方法

### 3. CompletionService ###

`Callable + Future` 虽然可以实现多个 task 并行执行，但是如果遇到前面的 task 执行较慢时需要阻塞等待前面的 task 执行完后面 task 才能取得结果。而 `CompletionService` 的主要功能就是：一边生成任务,一边获取任务的返回值。让两件事分开执行,任务之间不会互相阻塞，可以实现先执行完的先取结果，不再依赖任务顺序了

`ExecutorCompletionService` 类实现了 `CompletionService` 接口

> 使用 CompletionService 来优化上述的案例

public static void main(String[] args) throws InterruptedException, ExecutionException {
        
        ExecutorService executor = Executors.newFixedThreadPool(10);
        CompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
    
        Future<Integer> f1 = cs.submit(() -> {
        
            Thread.sleep(2000);
            return 1;
        });
        Future<Integer> f2 = cs.submit(() -> {
        
            Thread.sleep(1000);
            return 2;
        });
        for (int i = 0; i < 2; i++) {
        
            Integer integer = cs.take().get();
            System.out.println(integer);
        }
    }

`CompletionService` 接口：

public class ExecutorCompletionService<V> implements CompletionService<V> {
        
    	private final Executor executor;
        private final AbstractExecutorService aes;
        // 已完成的任务队列
        private final BlockingQueue<Future<V>> completionQueue;
    	
    	public Future<V> submit(Callable<V> task) {
        
    		// 将 Callable 转化为 FutureTask
            RunnableFuture<V> f = newTaskFor(task);
            // 会调用 FutureTask 类中的 run() 方法
            executor.execute(new QueueingFuture(f));
            return f;
        }
    	
    	// 将 Callable 转化为 FutureTask
    	private RunnableFuture<V> newTaskFor(Callable<V> task) {
        
            if (aes == null)
                return new FutureTask<V>(task);
            else
                return aes.newTaskFor(task);
        }
    	
    	// 内部类：任务完成后，将已完成的任务添加到队列中
    	private class QueueingFuture extends FutureTask<Void> {
        
            QueueingFuture(RunnableFuture<V> task) {
        
                super(task, null);
                this.task = task;
            }
            // done() 方法：FutureTask 中的 finishCompletion() 方法调用。这里，被子类重写
            protected void done() {
         completionQueue.add(task); }
            private final Future<V> task;
        }
    }

再来看看 `take()` 或者 `poll()`：都是操作 `completionQueue`

public Future<V> take() throws InterruptedException {
        
        return completionQueue.take();
    }
    
    public Future<V> poll() {
        
        return completionQueue.poll();
    }
    
    public Future<V> poll(long timeout, TimeUnit unit)
            throws InterruptedException {
        
        return completionQueue.poll(timeout, unit);
    }

### 4. CompletionStage ###

`CompletionStage` 是 Java 8 新增的接口，用于：**异步执行中的阶段处理**

> 场景：一个大的任务可以拆分成多个子任务，并且子任务之间有明显的先后顺序或者一个子任务依赖另一个子任务完成的结果时，那么 CompletionStage 是一个不错的选择

> CompletionStage 就是实现了将一个大任务分成若个子任务，这些子任务基于一定的并行、串行组合形成任务的不同阶段

查看其接口：

public interface CompletionStage<T> {
        
    
    	public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
    	//...
    	public CompletionStage<Void> runAfterBoth(CompletionStage<?> other, Runnable action);
    	//...
    	public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn);
    	//...
    }

`CompletionStage` 接口中方法很多，但不难发现有规律可循：基本都是由 then、apply、async、accept、run、combine、both、either、after、compose、when、handle 等关键词组合而成。这些关键字可以理解如下：

*  then：表示阶段先后顺序，即一个阶段等待另一个阶段完成
 *  apply：和 Function 一样，表示消费一个参数并提供一个结果
 *  async：异步标志，即阶段任务的执行相对于当前线程是同步还是异步
 *  accept：和 Consumer 一样，表示消费一个参数
 *  run：既不消费也不铲除，同 Runnable 接口含义
 *  combine：合并两个阶段结果并返回新的阶段
 *  both：表示二者条件都成立再做其它事
 *  either：表示二者之一条件成立再做其它事，对应 both
 *  after:表先后顺序，一个任务发生在另一个任务之后，和 then 相似
 *  compose：表示根据已有结果生成新的结果，同 Function，细看 compose 的参数和 thenApply 有区别，具体区别再下面陈述
 *  when：等同于 whenComplete，当前阶段正常完成或异常完成时执行 BiConsumer 动作
 *  handle：当前阶段正常完成或异常完成后触发一个 BiFunction 动作

### 5. CompletableFuture ###

`CompletableFuture` 实现了 `Future` 接口，并在此基础上进行了丰富的扩展，完美弥补了 `Future` 的局限性，同时`CompletableFuture` 实现了 `CompletionStage` 接口，对任务编排的能力

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
        
    	// 存储结果
    	volatile Object result;
    }

#### 5.1 常用方法 ####

**依赖关系**

*  `thenApply()`：把前面任务的执行结果，交给后面的 Function

CompletableFuture.supplyAsync(() -> 100).thenApply(x -> x * 100);

*  `thenCompose()`：用来连接两个有依赖关系的任务，结果由第二个任务返回

CompletableFuture.supplyAsync(() -> 100).thenCompose(x -> CompletableFuture.supplyAsync(() -> x * 100));

**and 集合关系**

*  `thenCombine()`：合并任务，有返回值

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 100);
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 200);
    Integer result = future1.thenCombine(future2, Integer::sum).get();

*  `thenAccepetBoth()`：两个任务执行完成后，将结果交给 thenAccepetBoth 处理，无返回值

future1.thenAcceptBoth(future2, (x, y) -> System.out.println(x + y));

*  `runAfterBoth()`：两个任务都执行完成后，执行下一步操作（Runnable 类型任务）。都是 **同步执行**

future1.runAfterBoth(future2, () -> System.out.println(Thread.currentThread().getName()));

**or 聚合关系**

*  `applyToEither()`：两个任务哪个执行的快，就使用哪一个结果，有返回值

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        
        Thread.sleep(1000);
        return 100;
    });
    CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
        
        Thread.sleep(500);
        return 200;
    });
    System.out.println(future1.applyToEither(future2, x -> x * 100).get());

*  `acceptEither()`：两个任务哪个执行的快，就消费哪一个结果，无返回值
 *  `runAfterEither()`：任意一个任务执行完成，进行下一步操作（Runnable 类型任务）

**并行执行**

*  `allOf()`：当所有给定的 CompletableFuture 完成时，返回一个新的 CompletableFuture
 *  `anyOf()`：当任何一个给定的CompletablFuture完成时，返回一个新的CompletableFuture

**异步操作**

*  `runAsync()`：
 *  `supplyAsync()`：

使用没有指定 Executor 的方法时，内部使用 `ForkJoinPool.commonPool()` 作为它的线程池执行异步代码。如果指定线程池，则使用指定的线程池运行。  
默认情况下 CompletableFuture 会使用公共的 ForkJoinPool 线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 `JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism` 来设置 ForkJoinPool 线程池的线程数）。如果所有 CompletableFuture 共享一个线程池，那么一旦有任务执行一些很慢的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，进而影响整个系统的性能。所以，强烈建议你要根据不同的业务类型创建不同的线程池，以避免互相干扰

public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        Runnable runnable = () -> System.out.println(Thread.currentThread().getName());
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(runnable, threadPool).exceptionally(e -> {
        
            System.out.println("当前线程执行失败");
            return null;
        });
        completableFuture.get(30, TimeUnit.SECONDS);
    }

**获取结果**

*  `get()`：
 *  `join()`：

**方法总结：**

![在这里插入图片描述][cbcd1a78f32e4566b3b3bb7bd1faea5c.png]

#### 5.2 场景 ####

著名数学家华罗庚先生在《统筹方法》这篇文章里介绍了一个烧水泡茶的例子，文中提到最优的工序应该是下面这样：  
![在这里插入图片描述][77c102f24e42413a82e3dfb9ae6fc79e.png]  
对于烧水泡茶这个程序，一种最优的分工方案：用两个线程 T1 和 T2 来完成烧水泡茶程序，T1 负责洗水壶、烧开水、泡茶这三道工序，T2 负责洗茶壶、洗茶杯、拿茶叶三道工序，其中 T1 在执行泡茶这道工序时需要等待 T2 完成拿茶叶的工序

**基于 Future 的实现**

public class T2 implements Callable<String> {
        
    
        @Override
        public String call() throws Exception {
        
            System.out.println("T2:洗茶壶...");
            TimeUnit.SECONDS.sleep(1);
    
            System.out.println("T2:洗茶杯...");
            TimeUnit.SECONDS.sleep(2);
    
            System.out.println("T2:拿茶叶...");
            TimeUnit.SECONDS.sleep(1);
            return "龙井";
        }
        
    }

public class T1 implements Callable<String> {
        
    
        private FutureTask<String> t2;
    
        public T1(FutureTask<String> t2) {
        
            this.t2 = t2;
        }
    
        @Override
        public String call() throws Exception {
        
            System.out.println("T1:洗水壶...");
            TimeUnit.SECONDS.sleep(1);
    
            System.out.println("T1:烧开水...");
            TimeUnit.SECONDS.sleep(15);
    
            // 获取 T2 的茶叶
            String chaYe = t2.get();
            System.out.println("T1:拿到茶叶:" + chaYe);
            System.out.println("T1:泡茶...");
            return "上茶:" + chaYe;
        }
    
    }

public static void main(String[] args) throws ExecutionException, InterruptedException {
        
        FutureTask<String> task2 = new FutureTask<>(new T2());
        FutureTask<String> task1 = new FutureTask<>(new T1(task2));
    
        new Thread(task1).start();
        new Thread(task2).start();
        System.out.println(task1.get());
    }

**基于 CompletableFuture 的实现**

public static void main(String[] args) throws ExecutionException, InterruptedException {
        
        //任务1：洗水壶->烧开水
        CompletableFuture<Void> task1 = CompletableFuture.runAsync(() -> {
        
            System.out.println("T1:洗水壶...");
            sleep(1, TimeUnit.SECONDS);
    
            System.out.println("T1:烧开水...");
            sleep(15, TimeUnit.SECONDS);
        });
        //任务2：洗茶壶->洗茶杯->拿茶叶
        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
        
            System.out.println("T2:洗茶壶...");
            sleep(1, TimeUnit.SECONDS);
    
            System.out.println("T2:洗茶杯...");
            sleep(2, TimeUnit.SECONDS);
    
            System.out.println("T2:拿茶叶...");
            sleep(1, TimeUnit.SECONDS);
            return "龙井";
        });
        //任务3：任务1和任务2完成后执行：泡茶
        CompletableFuture<String> f3 = task1.thenCombine(task2, (x, y) -> {
        
            System.out.println("T1:拿到茶叶:" + y);
            System.out.println("T1:泡茶...");
            return "上茶:" + y;
        });
        //等待任务3执行结果
        System.out.println(f3.join());
    }
    
    static void sleep(int t, TimeUnit u) {
        
        try {
        
            u.sleep(t);
        } catch (InterruptedException e) {
        
        }
    }

[Thread]: https://blog.csdn.net/sco5282/article/details/120809754?csdn_share_tail=%7B%22type%22:%22blog%22,%22rType%22:%22article%22,%22rId%22:%22120809754%22,%22source%22:%22sco5282%22%7D
[e89347c3d79d40d397ddb5fb9eeb2f84.png]: https://img-blog.csdnimg.cn/e89347c3d79d40d397ddb5fb9eeb2f84.png
[cbcd1a78f32e4566b3b3bb7bd1faea5c.png]: https://img-blog.csdnimg.cn/cbcd1a78f32e4566b3b3bb7bd1faea5c.png
[77c102f24e42413a82e3dfb9ae6fc79e.png]: https://img-blog.csdnimg.cn/77c102f24e42413a82e3dfb9ae6fc79e.png