【Java 多线程并发】 Future机制

Metadata

title: 【Java 多线程并发】 Future机制
date: 2023-07-05 09:54
tags:
  - 行动阶段/完成
  - 主题场景/程序
  - 笔记空间/KnowladgeSpace/ProgramSpace/BasicsSpace
  - 细化主题/Java
categories:
  - Java
keywords:
  - Java
description: 【Java 多线程并发】 Future机制

概述

核心思想

一个方法f，计算过程可能非常耗时，等待f返回，显然不明智。可以在调用f的时候，立马返回一个Future，可以通过Future这个数据结构去控制方法f的计算过程。

方法

  • get方法：获取计算结果（如果还没计算完，也是必须等待的）
  • cancel方法：还没计算完，可以取消计算过程
  • isDone方法：判断是否计算完
  • isCancelled方法：判断计算是否被取消

FutureTask

FutureTask是Future的具体实现。FutureTask实现了RunnableFuture接口。RunnableFuture接口又同时继承了Future 和 Runnable 接口。
FutureTask 提供了 Future 接口的基本实现，常用来封装 Callable 和 Runnable，具有取消任务、查看任务是否执行完成以及获取任务执行结果的方法。

state字段

  • volatile修饰的state字段；表示FutureTask当前所处的状态。
/**
     * Possible state transitions:
     * NEW -> COMPLETING -> NORMAL（创建到正常运行结束的状态变化轨迹）
     * NEW -> COMPLETING -> EXCEPTIONAL（创建到异常运行结束的状态变化轨迹）
     * NEW -> CANCELLED  （创建到取消的状态变化轨迹）
     * NEW -> INTERRUPTING -> INTERRUPTED（创建到中断结束的状态变化轨迹）
     */
    private volatile int state;
     // NEW 新建状态，表示这个 FutureTask还没有开始运行
    private static final int NEW          = 0;
       // COMPLETING 完成状态， 表示 FutureTask 任务已经计算完毕了
       // 但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
    private static final int COMPLETING   = 1;
       // FutureTask 任务完结，正常完成，没有发生异常
    private static final int NORMAL       = 2;
       // FutureTask 任务完结，因为发生异常。
    private static final int EXCEPTIONAL  = 3;
       // FutureTask 任务完结，因为取消任务
    private static final int CANCELLED    = 4;
       // FutureTask 任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求
    private static final int INTERRUPTING = 5;
       // FutureTask 任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求
    private static final int INTERRUPTED  = 6;

CompletableFuture

Future 不支持异步任务的编排组合、获取计算结果的 get() 方法为阻塞调用。

  • 观察者模式
  • 提供了更为好用和强大的 Future 特性
  • 函数式编程
  • 异步任务编排组合

CompletableFuture常用方法

  • 获取结果
    • public T get()
    • public T get(long timeout,TimeUnit unit)
    • public T join() —>和get一样的作用，只是不需要抛出异常
    • public T getNow(T valuelfAbsent) —>计算完成就返回正常值，否则返回备胎值（传入的参数），立即获取结果不阻塞
  • 主动触发计算
    • public boolean complete(T value) —->是否打断get方法立即返回括号值
  • 接受任务的处理结果，并消费处理，无返回结果
    • thenAccept
  • 对比补充
    • thenRun(Runnable runnable) :任务A执行完执行B，并且不需要A的结果
    • thenAccept(Consumer action): 任务A执行完执行B，B需要A的结果，但是任务B没有返回值
    • thenApply(Function fn): 任务A执行完执行B，B需要A的结果，同时任务B有返回值
  • 谁快用谁
    • applyToEither
  • 对计算结果进行合并
    • 两个CompletableStage任务都完成后，最终能把两个任务的结果一起交给thenCombine来处理
    • 先完成的先等着，等待其他分支任务

引入

先上一个场景：假如你突然想做饭，但是没有厨具，也没有食材。网上购买厨具比较方便，食材去超市买更放心。

实现分析：在快递员送厨具的期间，我们肯定不会闲着，可以去超市买食材。所以，在主线程里面另起一个子线程去网购厨具。

但是，子线程执行的结果是要返回厨具的，而run方法是没有返回值的。所以，这才是难点，需要好好考虑一下。

模拟代码1：

package test;
public class CommonCook {
    public static void main(String[] args) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        // 第一步 网购厨具
        OnlineShopping thread = new OnlineShopping();
        thread.start();
        thread.join();  // 保证厨具送到
 
        // 第二步 去超市购买食材
        Thread.sleep(2000);  // 模拟购买食材时间
        Shicai shicai = new Shicai();
        System.out.println("第二步：食材到位");
 
        // 第三步 用厨具烹饪食材
        System.out.println("第三步：开始展现厨艺");
        cook(thread.chuju, shicai);
        
        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
    }
    
    // 网购厨具线程
    static class OnlineShopping extends Thread {
        
        private Chuju chuju;
        @Override
        public void run() {
            System.out.println("第一步：下单");
            System.out.println("第一步：等待送货");
            try {
                Thread.sleep(5000);  // 模拟送货时间
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("第一步：快递送到");
            chuju = new Chuju();
        }
        
    }
    //  用厨具烹饪食材
    static void cook(Chuju chuju, Shicai shicai) {}
    
    // 厨具类
    static class Chuju {}
    
    // 食材类
    static class Shicai {}
}

运行结果：

  • 第一步：下单
  • 第一步：等待送货
  • 第一步：快递送到
  • 第二步：食材到位
  • 第三步：开始展现厨艺

总共用时7013ms

可以看到，多线程已经失去了意义。在厨具送到期间，我们不能干任何事。对应代码，就是调用join方法阻塞主线程。

有人问了，不阻塞主线程行不行？？？

不行！！！

从代码来看的话，run方法不执行完，属性chuju就没有被赋值，还是null，后面执行就会出现空指针异常。换句话说，没有厨具，怎么做饭。

Java现在的多线程机制，核心方法run是没有返回值的；如果要保存run方法里面的计算结果，必须等待run方法计算完，无论计算过程多么耗时。

面对这种尴尬的处境，程序员就会想：在子线程run方法计算的期间，能不能在主线程里面继续异步执行？？？

Where there is a will，there is a way！！！

这种想法的核心就是Future模式，下面先应用一下Java自己实现的Future模式。

模拟代码2：

package test;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
public class FutureCook {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        long startTime = System.currentTimeMillis();
        // 第一步 网购厨具
        Callable<Chuju> onlineShopping = new Callable<Chuju>() {
            @Override
            public Chuju call() throws Exception {
                System.out.println("第一步：下单");
                System.out.println("第一步：等待送货");
                Thread.sleep(5000);  // 模拟送货时间
                System.out.println("第一步：快递送到");
                return new Chuju();
            }
            
        };
        FutureTask<Chuju> task = new FutureTask<Chuju>(onlineShopping);
        new Thread(task).start();
        // 第二步 去超市购买食材
        Thread.sleep(2000);  // 模拟购买食材时间
        Shicai shicai = new Shicai();
        System.out.println("第二步：食材到位");
        // 第三步 用厨具烹饪食材
        if (!task.isDone()) {  // 联系快递员，询问是否到货
            System.out.println("第三步：厨具还没到，心情好就等着（心情不好就调用cancel方法取消订单）");
        }
        Chuju chuju = task.get();
        System.out.println("第三步：厨具到位，开始展现厨艺");
        cook(chuju, shicai);
        
        System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");
    }
    
    //  用厨具烹饪食材
    static void cook(Chuju chuju, Shicai shicai) {}
    
    // 厨具类
    static class Chuju {}
    
    // 食材类
    static class Shicai {}
}

运行结果：

第一步：下单
第一步：等待送货
第二步：食材到位
第三步：厨具还没到，心情好就等着（心情不好就调用cancel方法取消订单）
第一步：快递送到
第三步：厨具到位，开始展现厨艺

总共用时5005ms

可以看见，在快递员送厨具的期间，我们没有闲着，可以去买食材；而且我们知道厨具到没到，甚至可以在厨具没到的时候，取消订单不要了。

好神奇，有没有。

当然你可能会说模拟代码1中，只要join在cook(thread.chuju, shicai);就可以了，但是用future应该来说从编程角度更加自然。

比如模拟二的逻辑：

if (!task.isDone()) { // 联系快递员，询问是否到货
System.out.println("第三步：厨具还没到，心情好就等着（心情不好就调用cancel方法取消订单）");
}
Chuju chuju = task.get();

等价于模拟一：

thread.start();
// 第二步 去超市购买食材
Thread.sleep(2000); // 模拟购买食材时间
Shicai shicai = new Shicai();
System.out.println("第二步：食材到位");
thread.join(); // 保证厨具送到 ---------------join应该在购买食材开始后，而不是之前

用join这种写法也可以达到future的效果，但是使用Future不仅仅是更自然，而且确实是增加了灵活性，比如任务完成与否的判断，任务的取消等。这些是join不能做到的。

下面具体分析一下第二段代码：

1）把耗时的网购厨具逻辑，封装到了一个Callable的call方法里面。

public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

Callable接口可以看作是Runnable接口的补充，call方法带有返回值，并且可以抛出异常。

2）把Callable实例当作参数，生成一个FutureTask的对象，然后把这个对象当作一个Runnable，作为参数另起线程。

public class FutureTask<V> implements RunnableFuture<V>

public interface RunnableFuture<V> extends Runnable, Future<V>

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

这个继承体系中的核心接口是Future。Future的核心思想是：一个方法f，计算过程可能非常耗时，等待f返回，显然不明智。可以在调用f的时候，立马返回一个Future，可以通过Future这个数据结构去控制方法f的计算过程。

这里的控制包括：

get方法：获取计算结果（如果还没计算完，也是必须等待的）
cancel方法：还没计算完，可以取消计算过程
isDone方法：判断是否计算完
isCancelled方法：判断计算是否被取消

这些接口的设计很完美，FutureTask的实现注定不会简单，后面再说。

3）在第三步里面，调用了isDone方法查看状态，然后直接调用task.get方法获取厨具，不过这时还没送到，所以还是会等待3秒。对比第一段代码的执行结果，这里我们节省了2秒。这是因为在快递员送货期间，我们去超市购买食材，这两件事在同一时间段内异步执行。

为什么出现Future机制

常见的两种创建线程的方式。一种是直接继承Thread，另外一种就是实现Runnable接口。

这两种方式都有一个缺陷就是：在执行完任务之后无法获取执行结果。

从Java 1.5开始，就提供了Callable和Future，通过它们可以在任务执行完毕之后得到任务执行结果。

Future模式的核心思想是能够让主线程将原来需要同步等待的这段时间用来做其他的事情。（因为可以异步获得执行结果，所以不用一直同步等待去获得执行结果）

上图简单描述了不使用Future和使用Future的区别，不使用Future模式，主线程在invoke完一些耗时逻辑之后需要等待，这个耗时逻辑在实际应用中可能是一次RPC调用，可能是一个本地IO操作等。B图表达的是使用Future模式之后，我们主线程在invoke之后可以立即返回，去做其他的事情，回头再来看看刚才提交的invoke有没有结果。

Future 类有什么用？

Future 类是异步思想的典型运用，主要用在一些需要执行耗时任务的场景，避免程序一直原地等待耗时任务执行完成，执行效率太低。具体来说是这样的：当我们执行某一耗时的任务时，可以将这个耗时任务交给一个子线程去异步执行，同时我们可以干点其他事情，不用傻傻等待耗时任务执行完成。等我们的事情干完后，我们再通过 Future 类获取到耗时任务的执行结果。这样一来，程序的执行效率就明显提高了。

这其实就是多线程中经典的 Future 模式，你可以将其看作是一种设计模式，核心思想是异步调用，主要用在多线程领域，并非 Java 语言独有。

在 Java 中，Future 类只是一个泛型接口，位于 java.util.concurrent 包下，其中定义了 5 个方法，主要包括下面这 4 个功能：

  • 取消任务；
  • 判断任务是否被取消;
  • 判断任务是否已经执行完成;
  • 获取任务执行结果。
// V 代表了Future执行的任务返回值的类型
public interface Future<V> {
    // 取消任务执行
    // 成功取消返回 true，否则返回 false
    boolean cancel(boolean mayInterruptIfRunning);
    // 判断任务是否被取消
    boolean isCancelled();
    // 判断任务是否已经执行完成
    boolean isDone();
    // 获取任务执行结果
    V get() throws InterruptedException, ExecutionException;
    // 指定时间内没有返回计算结果就抛出 TimeOutException 异常
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutExceptio
}

简单理解就是：我有一个任务，提交给了 Future 来处理。任务执行期间我自己可以去做任何想做的事情。并且，在这期间我还可以取消任务以及获取任务的执行状态。一段时间之后，我就可以 Future 那里直接取出任务执行结果。

Future的相关类图

Future 接口

首先，我们需要清楚，Futrue是个接口。Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。

口定义行为，我们通过上图可以看到实现Future接口的子类会具有哪些行为：

  • 我们可以取消这个执行逻辑，如果这个逻辑已经正在执行，提供可选的参数来控制是否取消已经正在执行的逻辑。
  • 我们可以判断执行逻辑是否已经被取消。
  • 我们可以判断执行逻辑是否已经执行完成。
  • 我们可以获取执行逻辑的执行结果。
  • 我们可以允许在一定时间内去等待获取执行结果，如果超过这个时间，抛TimeoutException。

FutureTask 类

类图如下：

FutureTask是Future的具体实现。FutureTask实现了RunnableFuture接口。RunnableFuture接口又同时继承了Future 和 Runnable 接口。所以FutureTask既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。

FutureTask源码分析

state字段

volatile修饰的state字段；表示FutureTask当前所处的状态。可能的转换过程见注释。

/**
     * Possible state transitions:
     * NEW -> COMPLETING -> NORMAL（创建到正常运行结束的状态变化轨迹）
     * NEW -> COMPLETING -> EXCEPTIONAL（创建到异常运行结束的状态变化轨迹）
     * NEW -> CANCELLED  （创建到取消的状态变化轨迹）
     * NEW -> INTERRUPTING -> INTERRUPTED（创建到中断结束的状态变化轨迹）
     */
    private volatile int state;
     // NEW 新建状态，表示这个 FutureTask还没有开始运行
    private static final int NEW          = 0;
       // COMPLETING 完成状态， 表示 FutureTask 任务已经计算完毕了
       // 但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
    private static final int COMPLETING   = 1;
       // FutureTask 任务完结，正常完成，没有发生异常
    private static final int NORMAL       = 2;
       // FutureTask 任务完结，因为发生异常。
    private static final int EXCEPTIONAL  = 3;
       // FutureTask 任务完结，因为取消任务
    private static final int CANCELLED    = 4;
       // FutureTask 任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求
    private static final int INTERRUPTING = 5;
       // FutureTask 任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求
    private static final int INTERRUPTED  = 6;

其他变量

/** 任务 */
    private Callable<V> callable;
    /** 储存结果*/
    private Object outcome; // non-volatile, protected by state reads/writes
    /** 执行任务的线程*/
    private volatile Thread runner;
    /** get方法阻塞的线程队列 */
    private volatile WaitNode waiters;
    //FutureTask的内部类，get方法的等待队列
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }

CAS工具初始化

 
    // Unsafe mechanics
    private static final sun.misc.Unsafe UNSAFE;
    private static final long stateOffset;
    private static final long runnerOffset;
    private static final long waitersOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> k = FutureTask.class;
            stateOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("state"));
            runnerOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("runner"));
            waitersOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("waiters"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

这段代码是为了后面使用CAS而准备的。可以这么理解：

一个java对象可以看成是一段内存，各个字段都得按照一定的顺序放在这段内存里，同时考虑到对齐要求，可能这些字段不是连续放置的，用这个UNSAFE.objectFieldOffset（）方法能准确地告诉你某个字段相对于对象的起始内存地址的字节偏移量，因为是相对偏移量，所以它其实跟某个具体对象又没什么太大关系，跟class的定义和虚拟机的内存模型的实现细节更相关。

构造函数

FutureTask有两个构造函数：

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}
    
public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

这两个构造函数区别在于，如果使用第一个构造函数最后获取线程实行结果就是callable的执行的返回结果；而如果使用第二个构造函数那么最后获取线程实行结果就是参数中的result，接下来让我们看一下FutureTask的run方法。

同时两个构造函数都把当前状态设置为NEW。

jdk1.8和之前版本的区别

1.8:

get方法的逻辑很简单，如果call方法的执行过程已完成，就把结果给出去；如果未完成，就将当前线程挂起等待。awaitDone方法里面死循环的逻辑，推演几遍就能弄懂；它里面挂起线程的主要创新是定义了WaitNode类，来将多个等待线程组织成队列，这是与JDK6的实现最大的不同。

1.6：

JDK6的FutureTask的基本操作都是通过自己的内部类Sync来实现的，而Sync继承自AbstractQueuedSynchronizer这个出镜率极高的并发工具类

也就是说1.8自己定义了一个WaitNode类，来将多个等待线程组织成队列，在以前使用AQS来实现的

这个get方法里面处理等待线程队列的方式是调用了acquireSharedInterruptibly方法

Callable 和 Future 有什么关系？

我们可以通过 FutureTask 来理解 Callable 和 Future 之间的关系。

FutureTask 提供了 Future 接口的基本实现，常用来封装 Callable 和 Runnable，具有取消任务、查看任务是否执行完成以及获取任务执行结果的方法。ExecutorService.submit() 方法返回的其实就是 Future 的实现类 FutureTask 。

<T> Future<T> submit(Callable<T> task);
Future<?> submit(Runnable task);

FutureTask 不光实现了 Future接口，还实现了Runnable 接口，因此可以作为任务直接被线程执行。

FutureTask 有两个构造函数，可传入 Callable 或者 Runnable 对象。实际上，传入 Runnable 对象也会在方法内部转换为Callable 对象。

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;
}
public FutureTask(Runnable runnable, V result) {
    // 通过适配器RunnableAdapter来将Runnable对象runnable转换成Callable对象
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;
}

FutureTask相当于对Callable 进行了封装，管理着任务执行的情况，存储了 Callable 的 call 方法的任务执行结果。

CompletableFuture 类

Future 在实际使用过程中存在一些局限性，比如不支持异步任务的编排组合、获取计算结果的 get() 方法为阻塞调用。

Java 8 才被引入CompletableFuture 类可以解决Future 的这些缺陷。CompletableFuture 除了提供了更为好用和强大的 Future 特性之外，还提供了函数式编程、异步任务编排组合（可以将多个异步任务串联起来，组成一个完整的链式调用）等能力。

下面我们来简单看看 CompletableFuture 类的定义。

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {

}

可以看到，CompletableFuture 同时实现了 Future 和 CompletionStage 接口。

CompletionStage 接口描述了一个异步计算的阶段。很多计算可以分成多个阶段或步骤，此时可以通过它将所有步骤组合起来，形成异步计算的流水线。

CompletionStage 接口中的方法比较多，CompletableFuture 的函数式能力就是这个接口赋予的。从这个接口的方法参数你就可以发现其大量使用了 Java8 引入的函数式编程。

CompletableFuture对Future的改进

CompletableFuture为什么会出现

  1. get()方法在Future计算完成之前会一直处在阻塞状态下，阻塞的方式和异步编程的设计理念相违背。
  2. isDene()方法容易耗费cpu资源（cpu空转），
  3. 对于真正的异步处理我们希望是可以通过传入回调函数，在Future结束时自动调用该回调函数，这样，我们就不用等待结果

jdk8设计出CompletableFuture，CompletableFuture提供了一种观察者模式类似的机制，可以让任务执行完成后通知监听的一方。

CompletableFuture和CompletionStage介绍

类架构说明：

  • 接口CompletionStage
    • 代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段。
    • 一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发
  • 类CompletableFuture
    • 提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合CompletableFuture的方法
    • 它可能代表一个明确完成的Future，也可能代表一个完成阶段（CompletionStage），它支持在计算完成以后触发一些函数或执行某些动作

核心的四个静态方法，来创建一个异步任务

四个静态构造方法

对于上述Executor参数说明：若没有指定，则使用默认的ForkJoinPoolcommonPool（）作为它的线程池执行异步代码，如果指定线程池，则使用我们自定义的或者特别指定的线程池执行异步代码

/**
 * @author Guanghao Wei
 * @create 2023-04-10 12:16
 */
public class CompletableFutureBuildDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },executorService);

        System.out.println(completableFuture.get()); //null


        CompletableFuture<String> objectCompletableFuture = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello supplyAsync";
        },executorService);

        System.out.println(objectCompletableFuture.get());//hello supplyAsync

        executorService.shutdown();

    }
}

CompletableFuture减少阻塞和轮询，可以传入回调对象，当异步任务完成或者发生异常时，自动调用回调对象的回调方法。

/**
 * @author Guanghao Wei
 * @create 2023-04-10 12:28
 */
public class CompletableFutureUseDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "---come in");
            int result = ThreadLocalRandom.current().nextInt(10);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (result > 5) { //模拟产生异常情况
                int i = 10 / 0;
            }
            System.out.println("----------1秒钟后出结果" + result);
            return result;
        }, executorService).whenComplete((v, e) -> {
            if (e == null) {
                System.out.println("计算完成 更新系统" + v);
            }
        }).exceptionally(e -> {
            e.printStackTrace();
            System.out.println("异常情况：" + e.getCause() + " " + e.getMessage());
            return null;
        });
        System.out.println(Thread.currentThread().getName() + "先去完成其他任务");
        executorService.shutdown();
    }
}

/**
 * 无异常情况
 * pool-1-thread-1---come in
 * main先去完成其他任务
 * ----------1秒钟后出结果9
 * 计算完成 更新系统9
 */

/**
 * 有异常情况
 *pool-1-thread-1---come in
 * main先去完成其他任务
 * java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
 * 异常情况：java.lang.ArithmeticException: / by zero java.lang.ArithmeticException: / by zero
 */

CompletableFuture优点：

  • 异步任务结束时，会自动回调某个对象的方法
  • 主线程设置好回调后，不用关心异步任务的执行，异步任务之间可以顺序执行
  • 异步任务出错时，会自动回调某个对象的方法

CompletableFuture常用方法

获得结果和触发计算

  • 获取结果
    • public T get()
    • public T get(long timeout,TimeUnit unit)
    • public T join() —>和get一样的作用，只是不需要抛出异常
    • public T getNow(T valuelfAbsent) —>计算完成就返回正常值，否则返回备胎值（传入的参数），立即获取结果不阻塞
  • 主动触发计算
    • public boolean complete(T value) —->是否打断get方法立即返回括号值

对计算结果进行处理

  • thenApply —>计算结果存在依赖关系，这两个线程串行化—->由于存在依赖关系（当前步错，不走下一步），当前步骤有异常的话就叫停
  • handle —>计算结果存在依赖关系，这两个线程串行化—->有异常也可以往下走一步
/**
 * @author Guanghao Wei
 * @create 2023-04-10 13:43
 */
public class CompletableFutureApiDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 1;
        }, threadPool).thenApply(f -> {
            System.out.println("222");
            return f + 2;
        }).handle((f, e) -> {
            System.out.println("3333");
            int i=10/0;
            return f + 2;

//             thenApply(f -> {
//            System.out.println("3333");
//            return f + 2;
        }).whenComplete((v, e) -> {
            if (e == null) {
                System.out.println("----计算结果" + v);
            }
        }).exceptionally(e -> {
            e.printStackTrace();
            System.out.println(e.getCause());
            return null;
        });
        System.out.println(Thread.currentThread().getName() + "------主线程先去做其他事情");
    }
}

对计算结果进行消费

  • 接受任务的处理结果，并消费处理，无返回结果
    • thenAccept
/**
 * @author Guanghao Wei
 * @create 2023-04-10 13:59
 */
public class CompletableFutureApi2Demo {
    public static void main(String[] args) {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        CompletableFuture.supplyAsync(() -> {
            return 1;
        }, threadPool).thenApply(f -> {
            return f + 2;
        }).thenApply(f -> {
            return f + 2;
        }).thenAccept(r -> {
            System.out.println(r);//5
        });
    }
}
  • 对比补充
    • thenRun(Runnable runnable) :任务A执行完执行B，并且不需要A的结果
    • thenAccept(Consumer action): 任务A执行完执行B，B需要A的结果，但是任务B没有返回值
    • thenApply(Function fn): 任务A执行完执行B，B需要A的结果，同时任务B有返回值
/**
 * @author Guanghao Wei
 * @create 2023-04-10 13:59
 */
public class CompletableFutureApi2Demo {
    public static void main(String[] args) {
        System.out.println(CompletableFuture.supplyAsync(() -> "result").thenRun(() -> {}).join());//null
        System.out.println(CompletableFuture.supplyAsync(() -> "result").thenAccept(r -> System.out.println(r)).join());//result null
        System.out.println(CompletableFuture.supplyAsync(() -> "result").thenApply(f -> f + 2).join());//result2
    }
}

对计算速度进行选用

  • 谁快用谁
    • applyToEither
/**
 * @author Guanghao Wei
 * @create 2023-04-10 14:11
 * 可以合并写在一起，不必拆分
 */
public class CompletableFutureApiDemo {
    public static void main(String[] args) {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        CompletableFuture<String> playA = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("A come in");
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "playA";
        }, threadPool);


        CompletableFuture<String> playB = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("B come in");
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "playB";
        }, threadPool);

        CompletableFuture<String> result = playA.applyToEither(playB, f -> {
            return f + " is winner";
        });

        /**
         * A come in
         * B come in
         * main-----------winner:playA is winner
         */
        System.out.println(Thread.currentThread().getName() + "-----------winner:" + result.join());
    }
}

对计算结果进行合并

  • 两个CompletableStage任务都完成后，最终能把两个任务的结果一起交给thenCombine来处理
  • 先完成的先等着，等待其他分支任务
/**
 * @author Guanghao Wei
 * @create 2023-04-10 14:28
 * 可以合并写在一起，不必拆分
 */
public class CompletableFutureApi3Demo {
    public static void main(String[] args) {
        CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " 启动");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 10;
        });

        CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " 启动");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 20;
        });

        CompletableFuture<Integer> finalResult = completableFuture1.thenCombine(completableFuture2, (x, y) -> {
            System.out.println("----------开始两个结果合并");
            return x + y;
        });
        System.out.println(finalResult.join());

    }
}