异步编程 CompletableFuture

爱被打了一巴掌 2023-10-08 14:34 47阅读 0赞

#### 文章目录 ####

*  写在前面
 *  1、Future
 *   *  1.1、Future接口
     *  1.2、Future实例
     *  1.3、Future的缺点
 *  2、CompletableFuture
 *   *  2.1、CompletableFuture弥补了Future的缺点
     *  2.2、CompletableFuture类
     *   *  2.2.1、runAsync()、supplyAsync() 创建异步Future
         *  2.2.2、complete() 立即完成异步并返回结果
         *  2.2.3、thenApply()、thenApplyAsync() 改变/转换返回结果
         *  2.2.4、thenCompose()、thenCombine()、thenAcceptBoth()组合Future
         *  2.2.5、计算结果完成时的处理
         *   *  whenComplete() 完成时处理，无返回值
             *  handle()完成时处理，有返回值
             *  thenAccept() 只执行动作，不返回新的计算值
         *  2.2.6、Either()有一个任务完成就执行
         *  2.2.7、其他方法
 *  3、最后

## 写在前面 ##

## 1、Future ##

JDK 5引入了Future模式。可以用来做异步计算。

Future模式可以理解成：我有一个任务，提交给了Future，Future替我完成这个任务。期间我自己可以去做任何想做的事情。一段时间之后，我就便可以从Future那儿取出结果。

### 1.1、Future接口 ###

Future接口只有五个方法，如下：

public interface Future<V> {
 
 
 boolean cancel(boolean mayInterruptIfRunning); // 取消任务的执行。参数指定是否立即中断任务执行，或者等等任务结束 
 
 boolean isCancelled(); // 任务是否已经取消，任务正常完成前将其取消，则返回 true
 
 boolean isDone(); // 任务是否已经完成。需要注意的是如果任务正常终止、异常或取消，都将返回true
 	 
 
 V get() throws InterruptedException, ExecutionException; // 任务结束后得到结果
 
 V get(long timeout, TimeUnit unit) // 带超时时间的获得结果
 throws InterruptedException, ExecutionException, TimeoutException;
 
 	// InterruptedException 线程被中断异常
 	// ExecutionException 任务执行异常，如果任务被取消，还会抛出CancellationException
 }

> 比起future.get()，更推荐get (long timeout, TimeUnit unit) ，设置了超时，防止无限制的等待future的结果。

### 1.2、Future实例 ###

一般情况下，我们会结合Callable和Future一起使用，通过ExecutorService的submit方法执行Callable，并返回Future。

ExecutorService executor = Executors.newCachedThreadPool();
 
 Future<String> future = executor.submit(() -> {
 
 	// 这个Lambda实现的接口是callable，提交后立即执行，返回FutureTask实例
 	// submit()方法源码： <T> Future<T> submit(Callable<T> task);
 System.out.println("running task");
 Thread.sleep(10000);
 return "return task";
 });
 
 try {
 
 Thread.sleep(1000);
 } catch (InterruptedException e) {
 }
 
 		// 前面 Callable 在其他线程中运行着，这里可以做一些其他的事情
 System.out.println("do something else"); 
 
 try {
 
 // 等待 future 的执行结果，执行完毕之后打印出来
 System.out.println(future.get()); 
 } catch (InterruptedException e) {
 
 } catch (ExecutionException e) {
 
 } finally {
 
 executor.shutdown();
 }

### 1.3、Future的缺点 ###

可以获取异步执行结果，但无法得知Future什么时候完成。

*  如果使用阻塞，在future.get()那儿等待返回结果，就变成了同步操作。
 *  如果使用isDone()轮询判断Future是否完成，会耗费CPU的资源。

## 2、CompletableFuture ##

Java 8新增的CompletableFuture类，弥补了Future的缺点。

### 2.1、CompletableFuture弥补了Future的缺点 ###

*  能够将回调放到与任务不同的线程中执行。
 *  能将回调作为继续执行的同步函数，在与任务相同的线程中执行。
 *  能够将控制流分离到不同的事件处理器中。

> 也就是在异步的任务完成后，需要用其结果继续操作时，无需等待。可以直接通过thenAccept、thenApply、thenCompose等方式将前面异步处理的结果交给另外一个异步事件处理线程来处理。

### 2.2、CompletableFuture类 ###

CompletableFuture继承了Future、CompletionStage接口

#### 2.2.1、runAsync()、supplyAsync() 创建异步Future ####

CompletableFuture里面的静态工厂方法runAsync()、supplyAsync()

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
 
 
 	// 使用ForkJoinPool.commonPool()作为它的线程池执行异步代码。
 	public static CompletableFuture<Void> runAsync(Runnable runnable) {
 ...}
 	// 使用指定的thread pool执行异步代码。
 	public static CompletableFuture<Void> runAsync(Runnable runnable,
 Executor executor) {
 ...}
 
 
 // 使用ForkJoinPool.commonPool()作为它的线程池执行异步代码，异步操作有返回值
 public static CompletableFuture supplyAsync(Supplier supplier){
 ...}
 public static CompletableFuture supplyAsync(Supplier supplier,
 Executor executor) {
 ...}
 
 	
 }

如下实例：

runAsync() ，没有返回值

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
 
 System.out.println("Hello");
 });
 
 try {
 
 future.get();
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 
 System.out.println("CompletableFuture");

supplyAsync，有返回值

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 
 System.out.println("CompletableFuture");

#### 2.2.2、complete() 立即完成异步并返回结果 ####

CompletableFuture里面的complete()

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
 
 	// 完成异步执行，并返回future的结果
 	public boolean complete(T value){
 ...} 
 
 	// 异步执行不正常的结束
 	public boolean completeExceptionally(Throwable ex) {
 ...}
 }

如下实例：

future.get()在等待执行结果时，程序会一直block，调用complete(T value)会立即执行。  
complete(T value)只能调用一次，后续的重复调用会失效。

// 实例1
 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
 
 future.complete("World");
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 
 // 执行结果：World
 
 
 // 实例2
 // 如果future已经执行完毕能够返回结果，此时再调用complete(T t)则会无效。
 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
 
 try {
 
 Thread.sleep(5000);
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 
 future.complete("World");
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 
 // 执行结果：Hello
 
 
 // 实例3
 // 如果使用completeExceptionally(Throwable ex)则抛出一个异常，而不是一个成功的结果。
 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
 
 future.completeExceptionally(new Exception());
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 // 执行结果：java.util.concurrent.ExecutionException: java.lang.Exception...

#### 2.2.3、thenApply()、thenApplyAsync() 改变/转换返回结果 ####

可以改变或者转换返回结果的有如下几个方法：

// 入参fn：“Future的返回值”
 // 返回值：”Future的返回值“转换过后的值
 thenApply(Function<? super T,? extends U> fn)
 
 // 入参fn：使用ForkJoinPool的“Future的返回值”
 thenApplyAsync(Function<? super T,? extends U> fn)
 
 // 使用指定的线程池
 thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

实例

实例1：将结果拼接一个值后、再全部转为大写，再返回
 CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
 .thenApply(s -> s + " World").thenApply(String::toUpperCase);
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 // 执行结果：HELLO WORLD
 
 
 实例2：将结果转为Integer、再转为Double
 CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> "10")
 .thenApply(Integer::parseInt)
 .thenApply(i->i*10.0);
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 // 执行结果：100.0

#### 2.2.4、thenCompose()、thenCombine()、thenAcceptBoth()组合Future ####

线程都完成后进行组合的方法如下：  
（多个CompletableFuture间有先后顺序）

// 在异步操作完成的时候对异步操作的结果进行一些操作，并且仍然返回CompletableFuture类型。
 thenCompose(Function<? super T, ? extends CompletionStage> fn)
 // 在异步操作完成的时候对异步操作的结果进行一些操作，并且仍然返回CompletableFuture类型。使用ForkJoinPool。
 thenComposeAsync(Function<? super T, ? extends CompletionStage> fn)
 // 使用指定的线程池。
 thenComposeAsync(Function<? super T, ? extends CompletionStage> fn,Executor executor)

实例：

CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> "100")
 .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "100"))
 .thenCompose(s -> CompletableFuture.supplyAsync(() -> Double.parseDouble(s)));
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 // 执行结果：100100.0

线程都完成后进行组合的方法如下：  
（多个CompletableFuture并行执行）

// 返回CompletableFuture<Object>
 thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
 thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
 thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)
 
 
 // 返回CompletableFuture<Void>
 thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
 thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action)
 thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action, Executor executor)

实例：

实例1：thenCombine()
 
 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "100");
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 100);
 
 CompletableFuture<Double> future = future1.thenCombine(future2, (s, i) -> Double.parseDouble(s + i));
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 
 // 执行结果：100100.0
 
 
 
 实例2：thenAcceptBoth()
 
 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "100");
 CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 100);
 
 CompletableFuture<Void> future = future1.thenAcceptBoth(future2, (s, i) -> System.out.println(Double.parseDouble(s + i)));
 
 try {
 
 future.get();
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 
 // 执行结果：100100.0

#### 2.2.5、计算结果完成时的处理 ####

当CompletableFuture完成计算结果后，我们可能需要对结果进行一些处理。

##### whenComplete() 完成时处理，无返回值 #####

当CompletableFuture完成计算结果时对结果进行处理，或者当CompletableFuture产生异常的时候对异常进行处理。

有如下方法：

whenComplete(BiConsumer<? super T,? super Throwable> action)
 whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
 whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)

实例

CompletableFuture.supplyAsync(() -> "Hello")
                    .thenApply(s->s+" World")
                    .thenApply(s->s+ "\nThis is CompletableFuture demo")
                    .thenApply(String::toLowerCase)
                    .whenComplete((result, throwable) -> System.out.println(result));
    
    // 执行结果：
    // hello world
    // this is completablefuture demo

##### handle()完成时处理，有返回值 #####

当CompletableFuture完成计算结果或者抛出异常的时候，执行提供的fn

有如下方法：

handle(BiFunction<? super T, Throwable, ? extends U> fn)
 handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
 handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

实例：

CompletableFuture<Double> future = CompletableFuture.supplyAsync(() -> "100")
 .thenApply(s->s+"100")
 .handle((s, t) -> s != null ? Double.parseDouble(s) : 0);
 
 try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }
 
 // 执行结果：100100.0

> handle()相当于whenComplete()+转换。

`(s, t) -> s != null ? Double.parseDouble(s) : 0`这句代码分段如下  
![在这里插入图片描述][b0cddf962d00464eb4d997aa670a68d3.png]

##### thenAccept() 只执行动作，不返回新的计算值 #####

当CompletableFuture完成计算结果，只对结果执行Action，而不返回新的计算值

有如下方法：

thenAccept(Consumer<? super T> action)
 thenAcceptAsync(Consumer<? super T> action)
 thenAcceptAsync(Consumer<? super T> action, Executor executor

实例：

CompletableFuture.supplyAsync(() -> "Hello")
                    .thenApply(s->s+" World")
                    .thenApply(s->s+ "\nThis is CompletableFuture demo")
                    .thenApply(String::toLowerCase)
                    .thenAccept(System.out::print);
    
    // 执行结果：
    // hello world
    // this is completablefuture demo

#### 2.2.6、Either()有一个任务完成就执行 ####

Either 表示的是两个CompletableFuture，当其中任意一个CompletableFuture计算完成的时候就会执行。

有如下方法：

// 当任意一个CompletableFuture完成的时候，action这个消费者就会被执行。
 acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
 acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
 acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)

实例

Random random = new Random();
 
 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{
 
 
 try {
 
 Thread.sleep(random.nextInt(1000));
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 
 return "from future1";
 });
 
 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{
 
 
 try {
 
 Thread.sleep(random.nextInt(1000));
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 
 return "from future2";
 });
 
 CompletableFuture<Void> future = future1.acceptEither(future2,str->System.out.println("The future is "+str));
 
 try {
 
 future.get();
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }

执行结果：The future is from future1 或者 The future is from future2。  
因为future1和future2，执行的顺序是随机的。

applyToEither 跟 acceptEither 类似。只是要返回最后结果  
有如下方法：

// 当任意一个CompletableFuture完成的时候，fn会被执行，它的返回值会当作新的CompletableFuture的计算结果。
 applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn)
 applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn)
 applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor)

实例

Random random = new Random();
 
 CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{
 
 
 try {
 
 Thread.sleep(random.nextInt(1000));
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 
 return "from future1";
 });
 
 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{
 
 
 try {
 
 Thread.sleep(random.nextInt(1000));
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 }
 
 return "from future2";
 });
 
 CompletableFuture<String> future = future1.applyToEither(future2,str->"The future is "+str);
 	try {
 
 System.out.println(future.get());
 } catch (InterruptedException e) {
 
 e.printStackTrace();
 } catch (ExecutionException e) {
 
 e.printStackTrace();
 }

#### 2.2.7、其他方法 ####

// 在所有Future对象完成后结束，并返回一个future。
 allOf(CompletableFuture<?>... cfs)
 
 // 在任何一个Future对象结束后结束，返回第一个结束的future。
 anyOf(CompletableFuture<?>... cfs)
 
 // 只有当CompletableFuture抛出异常的时候，才会触发这个exceptionally的计算，调用function计算值。
 exceptionally(Function<Throwable,? extends T> fn)

allOf()实例:

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "tony");
 
 CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "cafei");
 
 CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "aaron");
 
 CompletableFuture.allOf(future1, future2, future3)
 .thenApply(v ->
 Stream.of(future1, future2, future3)
 .map(CompletableFuture::join)
 .collect(Collectors.joining(" ")))
 .thenAccept(System.out::print);

exceptionally()实例：

CompletableFuture.supplyAsync(() -> "hello world")
                    .thenApply(s -> {
        
                        s = null;
                        int length = s.length();
                        return length;
                    }).thenAccept(i -> System.out.println(i))
                    .exceptionally(t -> {
        
                        System.out.println("Unexpected error:" + t);
                        return null;
                    });
    // 执行结果：Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException

whenComplete也可以捕获异常

CompletableFuture.supplyAsync(() -> "hello world")
                    .thenApply(s -> {
        
                        s = null;
                        int length = s.length();
                        return length;
                    }).thenAccept(i -> System.out.println(i))
                    .whenComplete((result, throwable) -> {
        
    
                        if (throwable != null) {
        
                           System.out.println("Unexpected error:"+throwable);
                        } else {
        
                            System.out.println(result);
                        }
    
                    });
    // 执行结果：Unexpected error:java.util.concurrent.CompletionException: java.lang.NullPointerException

## 3、最后 ##

CompletableFuture具有链式风格，它不会造成堵塞，依靠fork/join框架来启动新的线程实现异步与并发。

[b0cddf962d00464eb4d997aa670a68d3.png]: https://img-blog.csdnimg.cn/b0cddf962d00464eb4d997aa670a68d3.png