CompletableFuture实现异步编排

2023-02-01 17:39:06 浏览数 (3)

前言

为什么需要异步执行？

场景：电商系统中获取一个完整的商品信息可能分为以下几步：①获取商品基本信息 ②获取商品图片信息 ③获取商品促销活动信息 ④获取商品各种类的基本信息 等操作，如果使用串行方式去执行这些操作，假设每个操作执行1s，那么用户看到完整的商品详情就需要4s的时间，如果使用并行方式执行这些操作，可能只需要1s就可以完成。所以这就是异步执行的好处。

JDK5的Future接口

Future接口用于代表异步计算的结果，通过Future接口提供的方法可以查看异步计算是否执行完成，或者等待执行结果并获取执行结果，同时还可以取消执行。

列举Future接口的方法：

  • get()：获取任务执行结果，如果任务还没完成则会阻塞等待直到任务执行完成。如果任务被取消则会抛出CancellationException异常，如果任务执行过程发生异常则会抛出ExecutionException异常，如果阻塞等待过程中被中断则会抛出InterruptedException异常。
  • get(long timeout,Timeunit unit)：带超时时间的get()方法，如果阻塞等待过程中超时则会抛出TimeoutException异常。
  • cancel()：用于取消异步任务的执行。如果异步任务已经完成或者已经被取消，或者由于某些原因不能取消，则会返回false。如果任务还没有被执行，则会返回true并且异步任务不会被执行。如果任务已经开始执行了但是还没有执行完成，若mayInterruptIfRunning为true，则会立即中断执行任务的线程并返回true，若mayInterruptIfRunning为false，则会返回true且不会中断任务执行线程。
  • isCanceled()：判断任务是否被取消，如果任务在结束(正常执行结束或者执行异常结束)前被取消则返回true，否则返回false。
  • isDone()：判断任务是否已经完成，如果完成则返回true，否则返回false。需要注意的是：任务执行过程中发生异常、任务被取消也属于任务已完成，也会返回true。

使用Future接口和Callable接口实现异步执行：

代码语言：javascript复制
public static void main(String[] args) {
	// 快速创建线程池
	ExecutorService executorService = Executors.newFixedThreadPool(4);
	// 获取商品基本信息（可以使用Lambda表达式简化Callable接口，这里为了便于观察不使用）
	Future<String> future1 = executorService.submit(new Callable<String>() {
		@Override
		public String call() throws Exception {
			return "获取到商品基本信息";
		}
	});
	// 获取商品图片信息
	Future<String> future2 = executorService.submit(new Callable<String>() {
		@Override
		public String call() throws Exception {
			return "获取商品图片信息";
		}
	});
	// 获取商品促销信息
	Future<String> future3 = executorService.submit(new Callable<String>() {
		@Override
		public String call() throws Exception {
			return "获取商品促销信息";
		}
	});
	// 获取商品各种类基本信息
	Future<String> future4 = executorService.submit(new Callable<String>() {
		@Override
		public String call() throws Exception {
			return "获取商品各种类基本信息";
		}
	});
        // 获取结果
	try {
		System.out.println(future1.get());
		System.out.println(future2.get());
		System.out.println(future3.get());
		System.out.println(future4.get());
	} catch (InterruptedException | ExecutionException e) {
		e.printStackTrace();
	}finally {
		executorService.shutdown();
	}
}
复制代码

既然Future可以实现异步执行并获取结果，为什么还会需要CompletableFuture?

简述一下Future接口的弊端：

  • 不支持手动完成
    • 当提交了一个任务，但是执行太慢了，通过其他路径已经获取到了任务结果，现在没法把这个任务结果通知到正在执行的线程，所以必须主动取消或者一直等待它执行完成。
  • 不支持进一步的非阻塞调用
    • 通过Future的get()方法会一直阻塞到任务完成，但是想在获取任务之后执行额外的任务，因为 Future 不支持回调函数，所以无法实现这个功能。
  • 不支持链式调用
    • 对于Future的执行结果，想继续传到下一个Future处理使用，从而形成一个链式的pipline调用，这在 Future中无法实现。
  • 不支持多个 Future 合并
    • 比如有10个Future并行执行，想在所有的Future运行完毕之后，执行某些函数，是无法通过Future实现的。
  • 不支持异常处理
    • Future的API没有任何的异常处理的api，所以在异步运行时，如果出了异常问题不好定位。

使用Future接口可以通过get()阻塞式获取结果或者通过轮询＋isDone()非阻塞式获取结果，但是前一种方法会阻塞，后一种会耗费CPU资源，所以JDK的Future接口实现异步执行对获取结果不太友好，所以在JDK8时推出了CompletableFuture实现异步编排。

CompletableFuture的使用

CompletableFuture概述

JDK8中新增加了一个包含50个方法左右的类CompletableFuture，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

代码语言：javascript复制
public class CompletableFuture<T> implements Future<T>, CompletionStage<T>
复制代码

CompletableFuture类实现了Future接口和CompletionStage接口，即除了可以使用Future接口的所有方法之外，CompletionStage<T>接口提供了更多方法来更好的实现异步编排，并且大量的使用了JDK8引入的函数式编程概念。后面会细致的介绍常用的API。

① 创建CompletableFuture的方式

  1. 使用new关键字创建
代码语言：javascript复制
// 无返回结果
CompletableFuture<String> completableFuture = new CompletableFuture<>();
// 已知返回结果
CompletableFuture<String> completableFuture = new CompletableFuture<>("result");
// 已知返回结果（底层其实也是带参数的构造器赋值）
CompletableFuture<String> completedFuture = CompletableFuture.completedFuture("result");
复制代码

创建一个返回结果类型为String的CompletableFuture，可以使用Future接口的get()方法获取该值（同样也会阻塞）。

可以使用无参构造器返回一个没有结果的CompletableFuture，也可以通过构造器的传参CompletableFuture设置好返回结果，或者使用CompletableFuture.completedFuture(U value)构造一个已知结果的CompletableFuture。

  1. 使用CompletableFuture类的静态工厂方法（常用）
  2. runAsync() 无返回值
代码语言：javascript复制
// 使用默认线程池
public static CompletableFuture<Void> runAsync(Runnable runnable)
// 使用自定义线程池（推荐）
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor) 
复制代码

runAsync()方法的参数是Runnable接口，这是一个函数式接口，不允许返回值。当需要异步操作且不关心返回结果的时候可以使用runAsync()方法。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        // 通过Lambda表达式实现Runnable接口
        CompletableFuture.runAsync(()-> System.out.println("获取商品基本信息成功"), executor).get();
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}
复制代码
  1. supplyAsync() 有返回值
代码语言：javascript复制
// 使用默认线程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
// 使用自定义线程池（推荐）
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor)
复制代码

supplyAsync()方法的参数是Supplier<U>供给型接口（无参有返回值），这也是一个函数式接口，U是返回结果值的类型。当需要异步操作且关心返回结果的时候,可以使用supplyAsync()方法。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
	// 快速创建线程池
	ExecutorService executor = Executors.newFixedThreadPool(4);
	try {
		// 通过Lambda表达式实现执行内容，并返回结果通过CompletableFuture接收
		CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
			System.out.println("获取商品信息成功");
			return "信息";
		}, executor);
		// 输出结果
		System.out.println(completableFuture.get());
	} catch (InterruptedException | ExecutionException e) {
		e.printStackTrace();
	}finally {
		executor.shutdown();
	}
}  
复制代码

关于第二个参数Executor executor说明

在没有指定第二个参数（即没有指定线程池）时，CompletableFuture直接使用默认的ForkJoinPool.commonPool()作为它的线程池执行异步代码。

在实际生产中会使用自定义的线程池来执行异步代码，具体可以参考另一篇文章深入理解线程池ThreadPoolExecutor - 掘金 (juejin.cn)，里面的第二节有生产中怎么创建自定义线程的例子，可以参考一下。

② 获得异步执行结果

  1. get() 阻塞式获取执行结果
代码语言：javascript复制
public T get() throws InterruptedException, ExecutionException
复制代码

该方法调用后如果任务还没完成则会阻塞等待直到任务执行完成。如果任务执行过程发生异常则会抛出ExecutionException异常，如果阻塞等待过程中被中断则会抛出InterruptedException异常。

  1. get(long timeout, TimeUnit unit) 带超时的阻塞式获取执行结果
代码语言：javascript复制
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
复制代码

该方法调用后如果如果任务还没完成则会阻塞等待直到任务执行完成或者超出timeout时间，如果阻塞等待过程中超时则会抛出TimeoutException异常。

  1. getNow(T valueIfAbsent) 立刻获取执行结果
代码语言：javascript复制
public T getNow(T valueIfAbsent)
复制代码

该方法调用后，会立刻获取结果不会阻塞等待。如果任务完成则直接返回执行完成后的结果，如果任务没有完成，则返回调用方法时传入的参数valueIfAbsent值。

  1. join() 不抛异常的阻塞时获取执行结果
代码语言：javascript复制
public T join()
复制代码

该方法和get()方法作用一样，只是不会抛出异常。

  1. complete(T value) 主动触发计算，返回异步是否执行完毕
代码语言：javascript复制
public boolean complete(T value)
复制代码

该方法调用后，会主动触发计算结果，如果此时异步执行并没有完成（此时boolean值返回true），则通过get()拿到的数据会是complete()设置的参数value值，如果此时异步执行已经完成（此时boolean值返回false），则通过get()拿到的就是执行完成的结果。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        // 通过Lambda表达式实现执行内容，并返回结果通过CompletableFuture接收
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            // 休眠2秒，使得异步执行变慢，会导致主动触发计算先执行，此时返回的get就是555
            try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }
            return 666;
        }, executor);
        // 主动触发计算，判断异步执行是否完成
        System.out.println(completableFuture.complete(555));
        // 输出结果
        System.out.println(completableFuture.get());
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}

/**
输出结果:
    true
    555
**/
复制代码

③ 对执行结果进行处理

whenComplete 等待前面任务执行完再执行当前处理

代码语言：javascript复制
public CompletableFuture<T> whenComplete(
        BiConsumer<? super T, ? super Throwable> action)
复制代码

在创建好的初始任务或者是上一个任务后通过链式调用该方法，会在之前任务执行完成后继续执行whenComplete里的内容（whenComplete传入的action只是对之前任务的结果进行处理），即使用该方法可以避免前面说到的Future接口的问题，不再需要通过阻塞或者轮询的方式去获取结果，而是通过调用该方法等任务执行完毕自动调用。

该方法的参数为BiConsumer<? super T, ? super Throwable> action消费者接口，可以接收两个参数，一个是任务执行完的结果，一个是执行任务时的异常。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        CompletableFuture.supplyAsync(() -> 666, executor)
                .whenComplete((res, ex) -> System.out.println("任务执行完毕，结果为"   res   " 异常为"   ex)
                );
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}

/**
输出结果:
    任务执行完毕，结果为666 异常为null
**/
复制代码

除了上述的方法外，还有一些类似的方法如XXXAsync()或者是XXXAsync(XX,Executor executor),对于这些方法，这里统一说明，后续文章中将不会再列举

代码语言：javascript复制
public CompletableFuture<T> whenCompleteAsync(
        BiConsumer<? super T, ? super Throwable> action)

public CompletableFuture<T> whenCompleteAsync(
        BiConsumer<? super T, ? super Throwable> action, Executor executor)
复制代码

XXXAsync()：表示上一个任务执行完成后，不会再使用之前任务中的线程，而是重新使用从默认线程（ForkJoinPool 线程池）中重新获取新的线程执行当前任务。

XXXAsync(XX,Executor executor)：表示不会沿用之前任务的线程，而是使用自己第二个参数指定的线程池重新获取线程执行当前任务。

④ 对执行结果进行消费

  1. thenRun 前面任务执行完后执行当前任务，不关心前面任务的结果，也没返回值
代码语言：javascript复制
public CompletableFuture<Void> thenRun(Runnable action)
复制代码

CompletableFuture.supplyAsync(actionA).thenRun(actionB)像这样链式调用该方法表示：执行任务A完成后接着执行任务B，但是任务B不需要A的结果，并且执行完任务B也不会返回结果。

thenRun(Runnable action)的参数为Runnable接口即没有传入参数。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
	// 快速创建线程池
	ExecutorService executor = Executors.newFixedThreadPool(4);
	try {
		CompletableFuture.supplyAsync(() -> 666, executor)
                    .thenRun(() -> System.out.println("我都没有参数怎么拿到之前的结果，我也没有返回值。")
                );
	} catch (Exception e) {
		e.printStackTrace();
	}finally {
		executor.shutdown();
	}
}

/**
输出结果:
    我都没有参数怎么拿到之前的结果，我也没有返回值。
**/
复制代码
  1. thenAccept 前面任务执行完后执行当前任务，消费前面的结果，没有返回值
代码语言：javascript复制
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
复制代码

CompletableFuture.supplyAsync(actionA).thenRun(actionB)像这样链式调用该方法表示：执行任务A完成后接着执行任务B，而且任务B需要A的结果，但是执行完任务B不会返回结果。

thenAccept(Consumer<? super T> action)的参数为消费者接口，即可以传入一个参数，该参数为上一个任务的执行结果。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        CompletableFuture.supplyAsync(() -> 666, executor)
                .thenAccept((res) -> System.out.println("我能拿到上一个的结果"   res   ",但是我没法传出去。")
                );
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}

/**
输出结果:
    我能拿到上一个的结果666,但是我没法传出去。
**/
复制代码
  1. thenApply 前面任务执行完后执行当前任务，消费前面的结果，具有返回值
代码语言：javascript复制
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
复制代码

CompletableFuture.supplyAsync(actionA).thenRun(actionB)像这样链式调用该方法表示：执行任务A完成后接着执行任务B，而且任务B需要A的结果，并且执行完任务B需要有返回结果。

thenApply(Function<? super T,? extends U> fn)的参数为函数式接口，即可以传入一个参数类型为T，该参数是上一个任务的执行结果，并且函数式接口需要有返回值,类型为U。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        CompletableFuture.supplyAsync(() -> 666, executor)
                .thenApply((res) -> {
                        System.out.println("我能拿到上一个的结果"   res   "并且我要将结果传出去");
                        return res;
                    }
                ).whenComplete((res, ex) -> System.out.println("结果"   res));
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}
/**
输出结果:
    我能拿到上一个的结果666并且我要将结果传出去
    结果666
**/
复制代码

⑤ 异常处理

  1. exceptionally 异常捕获，只消费前面任务中出现的异常信息，具有返回值
代码语言：javascript复制
public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)
复制代码

可以通过链式调用该方法来获取异常信息，并且具有返回值。如果某一个任务出现异常被exceptionally捕获到则剩余的任务将不会再执行。类似于Java异常处理的catch。

exceptionally(Function<Throwable, ? extends T> fn)的参数是函数式接口，具有一个参数以及返回值，该参数为前面任务的异常信息。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        CompletableFuture.supplyAsync(() -> {
                    if (Math.random() < 0.5) throw new RuntimeException("error");
                    return 666;
                }, executor)
                .thenApply((res) -> {
                    System.out.println("不出现异常，结果为"   res);
                    return res;
                }).exceptionally((ex) -> {
                    ex.printStackTrace();
                    return -1;
                });
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}

/**
输出结果:
// 这是不出现异常的情况
不出现异常，结果为666

// 这是出现异常的情况
java.util.concurrent.CompletionException: java.lang.RuntimeException: error
        at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)
        at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)
        at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1702)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.RuntimeException: error
        at com.xqsr.review.thread.ThreadTest.lambda$main$0(ThreadTest.java:15)
        at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
        ... 3 more
**/
复制代码
  1. handle 异常处理，消费前面的结果及异常信息，具有返回值，不会中断后续任务
代码语言：javascript复制
public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
复制代码

可以通过链式调用该方法可以跟thenApply()一样可以消费前面任务的结果并完成自己任务内容，并且具有返回值。不同之处在于出现异常也可以接着往下执行，根据异常参数做进一步处理。

handle(BiFunction<? super T, Throwable, ? extends U> fn)的参数是消费者接口，一个参数是任务执行结果，一个是异常信息，并且具有返回值。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        CompletableFuture.supplyAsync(() -> 666, executor)
                .thenApply((res) -> {
                    if (Math.random() < 0.5) throw new RuntimeException("error");
                    return res;
                }).handle((res, ex) -> {
                    System.out.println("结果"   res   "(null表示之前出现异常导致结果无法传过来)");
                    return res == null ? -1 : res;
                }).thenApply((res) -> {
                    System.out.println("结果为"   res   "(-1表示之前出现异常，经过handler使得结果处理成-1)");
                    return res;
                }).exceptionally((ex) -> {
                    ex.printStackTrace();
                    return -1;
                });
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}

/**
输出结果:
// 这是不出现异常的情况
结果666(null表示之前出现异常导致结果无法传过来)
结果为666(-1表示之前出现异常，经过handler使得结果处理成-1)

// 这是出现异常的情况
结果null(null表示之前出现异常导致结果无法传过来)
结果为-1(-1表示之前出现异常，经过handler使得结果处理成-1)
**/
复制代码

可以看到通过handle类似于Java异常处理的finally，出现异常并不会像使用exceptionally那样中断后续的任务，而是继续执行，可以通过handle为之前出现异常无法获得的结果重新赋值（根据业务需求设置安全值之类的）。

⑥ 两组任务按顺序执行

thenCompose 实现两组任务按前后顺序执行

代码语言：javascript复制
public <U> CompletableFuture<U> thenCompose(
    Function<? super T, ? extends CompletionStage<U>> fn)
复制代码

A.thenCompose(B)相当于任务A要排在任务B前面，即顺序的执行任务A、任务B。该方法的参数是函数式接口，函数式接口的参数是调用者的执行结果，返回值是另一个任务B。

代码语言：javascript复制
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        CompletableFuture<Integer> actionA = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务A先执行结果为666");
            return 666;
        }, executor);

        actionA.thenCompose((res) ->  CompletableFuture.supplyAsync(() -> {
            System.out.println("任务B后执行结果加上333");
            return 333   res;
        })).whenComplete((res, ex) -> System.out.println(res));
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}

/**
输出结果:
    任务A先执行结果为666
    任务B后执行结果加上333
    999
**/
复制代码

⑦ 两组任务谁快用谁

applyToEither 比较两组任务执行速度，谁快消费谁的执行结果

代码语言：javascript复制
public <U> CompletableFuture<U> applyToEither(
        CompletionStage<? extends T> other, Function<? super T, U> fn)
复制代码

该方法用于比较两组任务的执行速度，谁先执行完就用谁的执行结果。

传入参数说明：第一个参数传入的是另一个任务的执行内容，第二个参数传入的是最终这两个任务谁快返回谁的结果，并通过当前函数式接口进行接收和处理（使用函数式接口，有参且有返回值）。

代码语言：javascript复制
// 例子
public static void main(String[] args) {
    // 快速创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(4);
    try {
        CompletableFuture<Integer> actionA = CompletableFuture.supplyAsync(() -> {
            try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }
            System.out.println("任务A等待久一点，执行结果为555");
            return 555;
        }, executor);

        actionA.applyToEither(CompletableFuture.supplyAsync(() -> {
            System.out.println("任务B很快，执行结果为666");
            return 666;
        }), (res) -> {
            System.out.println("最终使用的执行结果为"   res);
            return res;
        });
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        executor.shutdown();
    }
}
/**
输出结果:
    任务B很快，执行结果为666
    最终使用的执行结果为666
    任务A等待久一点，执行结果为555
**/
复制代码

除了applyToEither对任务最终结果进行获取并消费，并且具有返回值的方法外，还有两个类似的方法。

代码语言：javascript复制
// 这个方法效果和上面的一样，比谁快拿谁的结果，不同的是这个方法只消费不具有返回值
public CompletableFuture<Void> acceptEither(
        CompletionStage<? extends T> other, Consumer<? super T> action)
复制代码
代码语言：javascript复制
// 这个方法效果和上面的一样，比谁快拿谁的结果，不同的是这个方法不消费结果也不具有返回值
public CompletableFuture<Void> runAfterEither(
        CompletionStage<?> other, Runnable action)
action extends handle super 异步

1 人点赞