国外 网站页面,电影网站源码access,站群cms系统,网络营销策划书4000字CompletableFuture 是由Java8引入的#xff0c;这让我们编写清晰可读的异步代码变得更加容易#xff0c;该类功能比Future 更加强大。 什么是 CompletableFuture
在Java中CompletableFuture用于异步编程#xff0c;异步通常意味着非阻塞#xff0c;运行任务单独的线程这让我们编写清晰可读的异步代码变得更加容易该类功能比Future 更加强大。 什么是 CompletableFuture
在Java中CompletableFuture用于异步编程异步通常意味着非阻塞运行任务单独的线程与主线程隔离。并且通过回调可以在主线程中得到异步任务的执行状态是否完成和异常等信息。
通过这种方式主线程不会被阻塞不需要一直等到子线程完成。主线程可以并行的执行其他任务。使用这种并行方式可以极大的提高程序的性能。 为什么要引入 CompletableFuture
一些业务场景我们需要使用多线程异步执行任务加快任务执行速度所以 JDK5 新增了 Future 接口用于描述一个异步计算的结果。
虽然 Future 以及相关使用方法提供了异步执行任务的能力但是对于结果的获取却是很不方便我们必须使用 Future.get 的方式阻塞调用线程或者使用轮询方式判断 Future.isDone 任务是否结束再获取结果。
FutureString future executor.submit(()-{Thread.sleep(2000);return hello world;
});// 获取结果
System.out.println(future.get());从上面的形式看来不能及时地得到计算结果所以要实现真正的异步上述这样是完全不够的。
若需要更强大的异步处理能力单纯使用 Future 接口或者 FutureTask 类并不能很好地完成以下业务场景
将两个异步计算合并为一个这两个异步计算之间相互独立同时第二个又依赖于第一个的结果等待Future集合种的所有任务都完成仅等待Future集合种最快结束的任务完成有可能因为他们试图通过不同的方式计算同一个值并返回它的结果通过编程方式完成一个Future任务的执行即以手工设定异步操作结果的方式应对Future的完成时间即当Future的完成时间完成时会收到通知并能使用Future的计算结果进行下一步的的操作不只是简单地阻塞等待操作的结果。
所以JDK 8.0新增了CompletableFuture 来解决上述这些痛点。 Future vs CompletableFuture
CompletableFuture 是 Future API的扩展。
Future 被用于作为一个异步计算结果的引用。提供一个 isDone() 方法来检查计算任务是否完成。当任务完成时get() 方法用来接收计算任务的结果。
Future 的局限性
不能手动完成
当你写了一个函数用于通过一个远程API获取一个电子商务产品最新价格。因为这个 API 太耗时你把它允许在一个独立的线程中并且从你的函数中返回一个 Future。现在假设这个API服务宕机了这时你想通过该产品的最新缓存价格手工完成这个Future 。你会发现无法这样做。
Future 的结果在非阻塞的情况下不能执行更进一步的操作
Future 不会通知你它已经完成了它提供了一个阻塞的 get() 方法通知你结果。你无法给 Future 植入一个回调函数当 Future 结果可用的时候用该回调函数自动的调用 Future 的结果。
多个 Future 不能串联在一起组成链式调用
有时候你需要执行一个长时间运行的计算任务并且当计算任务完成的时候你需要把它的计算结果发送给另外一个长时间运行的计算任务等等。你会发现你无法使用 Future 创建这样的一个工作流。
不能组合多个 Future 的结果
假设你有10个不同的Future你想并行的运行然后在它们运行未完成后运行一些函数。你会发现你也无法使用 Future 这样做。
没有异常处理
Future API 没有任务的异常处理结构居然有如此多的限制幸好我们有CompletableFuture你可以使用 CompletableFuture 达到以上所有目的。 CompletableFuture 实现了 Future 和 CompletionStage 接口并且提供了许多关于创建链式调用和组合多个 Future 的便利方法集而且有广泛的异常处理支持。 CompletableFuture的应用场景
执行比较耗时的操作时尤其是那些依赖一个或多个远程服务的操作使用异步任务可以改善程序的性能加快程序的响应速度。使用CompletableFuture类它提供了异常管理的机制让你有机会抛出、管理异步任务执行种发生的异常。如果这些异步任务之间相互独立或者他们之间的的某一些的结果是另一些的输入你可以讲这些异步任务构造或合并成一个。 CompletableFuture设计思想
CompletableFuture 按照类似“观察者模式”的设计思想原理分析可以从“观察者”和“被观察者”两个方面着手。 由于回调种类多但结构差异不大所以这里单以一元依赖中的thenApply为例不再枚举全部回调类型如下图所示 被观察者
每个CompletableFuture都可以被看作一个被观察者其内部有一个Completion类型的链表成员变量stack用来存储注册到其中的所有观察者。当被观察者执行完成后会弹栈stack属性依次通知注册到其中的观察者。上面例子中步骤fn2就是作为观察者被封装在UniApply中。被观察者CF中的result属性用来存储返回结果数据。这里可能是一次RPC调用的返回值也可能是任意对象在上面的例子中对应步骤fn1的执行结果。
观察者
CompletableFuture支持很多回调方法例如thenAccept、thenApply、exceptionally等这些方法接收一个函数类型的参数f生成一个Completion类型的对象即观察者并将入参函数f赋值给Completion的成员变量fn然后检查当前CF是否已处于完成状态即result!null如果已完成直接触发fn否则将观察者Completion加入到CF的观察者链stack中再次尝试触发如果被观察者未执行完则其执行完毕之后通知触发。
观察者中的dep属性指向其对应的CompletableFuture在上面的例子中dep指向CF2。观察者中的src属性指向其依赖的CompletableFuture在上面的例子中src指向CF1。观察者Completion中的fn属性用来存储具体的等待被回调的函数。这里需要注意的是不同的回调方法thenAccept、thenApply、exceptionally等接收的函数类型也不同即fn的类型有很多种在上面的例子中fn指向fn2。 CompletableFuture核心功能
CompletableFuture的功能主要体现在它的CompletionStage,如下图所示 CompletionStage接口定义了任务编排的方法执行某一阶段可以向下执行后续阶段可以实现如下功能
转换thenCompose组合thenCombine消费thenAccept运行thenRun带返回的消费thenApply
具体其他功能大家可以根据需求自行查看。 CompletableFuture创建使用
创建CompletableFuture对象提供了四个静态方法用来创建CompletableFuture对象
public static CompletableFutureVoid runAsync(Runnable runnable)
public static CompletableFutureVoid runAsync(Runnable runnable, Executor executor)
public static U CompletableFutureU supplyAsync(SupplierU supplier)
public static U CompletableFutureU supplyAsync(SupplierU supplier, Executor executor)
Asynsc 表示异步而 supplyAsync 与 runAsync 不同在与前者异步返回一个结果后者是 void。第二个函数第二个参数表示是用我们自己创建的线程池否则采用默认的 ForkJoinPool.commonPool() 作为它的线程池.其中Supplier是一个函数式接口代表是一个生成者的意思传入0个参数返回一个结果。
CompletableFutureString future CompletableFuture.supplyAsync(()-{return hello world;
});//阻塞的获取结果 helllo world
System.out.println(future.get());CompletableFuture用法详解
没有返回值的异步任务
如果你想异步的运行一个后台任务并且不需要任务返回结果就可以使用 runAsync()。
runAsync() 返回一个新的 CompletableFuture它在运行给定操作后由在 ForkJoinPool.commonPool() 运行的任务异步完成。
/*** 没有返回值的异步任务*/
Test
public void runAsync() throws Exception {CompletableFutureVoid future CompletableFuture.runAsync(() - {log.info(Current thread name: {}, Thread.currentThread().getName());// 模拟长时间运行的作业try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {throw new IllegalStateException(e);};});// 主线程阻塞future.get();System.out.println(主线程结束);
}
有返回值的异步任务
当运行一个异步任务并且需要有返回结果时就可以使用 supplyAsync()。
CompletableFuture.supplyAsync() 它持有 supplierT 并且返回 CompletableFutureTT 是通过调用传入的 supplier 取得的值的类型。
CompletableFutureString future CompletableFuture.supplyAsync(new SupplierString() {Overridepublic String get() {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new IllegalStateException(e);}return Success;}
});System.out.println(future.get());SupplierT 是一个简单的函数式接口表示 supplier 的结果。它有一个 get()该方法可以写入你的后台任务中并且返回结果。
还可以使用 lambda 表达式使得上面的示例更加简明
/*** 有返回值的异步任务*/
Test
public void supplyAsync() throws Exception {CompletableFutureString future CompletableFuture.supplyAsync(() - {log.info(Current thread name: {}, Thread.currentThread().getName());// 模拟长时间运行的作业try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {throw new IllegalStateException(e);};return Success;});// 主线程阻塞System.out.println(future.get());
}
上述 runAsync() 和 supplyAsync() 都是在单独的线程中执行他们的任务但在实际业务中我们不会只创建一个线程。
自定义线程池执行方法
CompletableFuture 可以从全局的 ForkJoinPool.commonPool() 获得一个线程中执行这些任务。但也可以创建一个线程池并传给 runAsync() 和 supplyAsync() 来让他们从线程池中获取一个线程执行它们的任务。
CompletableFuture API 的所有方法都有两个变体-一个接受Executor作为参数另一个不这样 static CompletableFutureVoid runAsync(Runnable runnable)
static CompletableFutureVoid runAsync(Runnable runnable, Executor executor)
static U CompletableFutureU supplyAsync(SupplierU supplier)
static U CompletableFutureU supplyAsync(SupplierU supplier, Executor executor)
创建一个线程池并传递给其中一个方法
Executor executor Executors.newFixedThreadPool(10);
CompletableFutureString future CompletableFuture.supplyAsync(() - {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new IllegalStateException(e);}return Success;
}, executor);
线程串行化
由于 CompletableFuture.get() 方法是阻塞的它会一直等到 Future 完成并且在完成后返回结果。但是这是我们想要的吗对于构建异步系统我们应该附上一个回调给 CompletableFuture当 Future 完成的时候自动的获取结果。
如果不想等待结果返回就可以把需要等待 Future 完成执行的逻辑写入到回调函数中。
可以使用 thenApply()、thenAccept() 、thenRun() 回调给 CompletableFuture。
thenApply()
当一个线程依赖另一个线程时可以使用 thenApply() 来把这两个线程串行化。
thenApply可以使用 thenApply() 处理和改变 CompletableFuture 的结果。
/*** 使用 thenApply() 处理和改变CompletableFuture的结果** throws Exception*/
Test
public void thenApply1() throws Exception {CompletableFutureString future CompletableFuture.supplyAsync(() - {return Java;}).thenApply(o - {return Hello o;});System.out.println(future.get());
}
thenAccept()
如果你不想从你的回调函数中返回任何东西仅仅想在 Future 完成后运行一些代码片段你可以使用 thenAccept() 和 thenRun()这些方法经常在调用链的最末端的最后一个回调函数中使用。
thenAccept消费处理结果接收任务的处理结果并消费处理无返回结果。
Test
public void thenAccept() throws Exception {CompletableFutureVoid future CompletableFuture.supplyAsync(() - {// 模拟长时间运行的作业try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {throw new IllegalStateException(e);}return Success;}).thenAccept(o - {if (Success.equals(o)) {// 模拟长时间运行的作业try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {throw new IllegalStateException(e);}}});future.get();log.info();System.out.println(结束.);
}
thenRun()
thenRun() 不能访 Future 的结果它持有一个 Runnable 返回 CompletableFuture
Test
public void thenRun() throws Exception {CompletableFutureVoid future CompletableFuture.supplyAsync(() - {// 模拟长时间运行的作业try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {throw new IllegalStateException(e);}return Success;}).thenRun(() - {// 作业完成后执行一些代码片段System.out.println(thenRun 执行一些代码片段);});future.get();
}
结果合并
thenCompose
使用 thenCompose() 合并两个有依赖关系的 CompletableFutures 的执行结果。
private static Integer num 10;Test
public void thenCompose() throws Exception {//第一步加 10CompletableFutureInteger future CompletableFuture.supplyAsync(() - {System.out.println(让num10;任务开始);num 10;return num;});//合并CompletableFutureInteger future1 future.thenCompose(i -//再来一个 CompletableFutureCompletableFuture.supplyAsync(() - i 1));System.out.println(future.get());System.out.println(future1.get());
}thenCombine()
使用 thenCombine() 组合两个独立的 future。当两个独立的 Future 都完成的时候使用 thenCombine() 用来做一些事情。
Test
public void thenCompose() throws Exception {// 长方形:S长*宽//第一步加 10CompletableFutureInteger lengthFuture CompletableFuture.supplyAsync(() - {return 50;});CompletableFutureInteger widthFuture CompletableFuture.supplyAsync(() - {return 30;});CompletableFutureInteger combinedFuture lengthFuture.thenCombine(widthFuture, (t1, t2) - {System.out.println(t1);System.out.println(t2);return t1 * t2;});System.out.println(combinedFuture.get());
}
合并多个任务的结果
使用 thenCompose() 和 thenCombine() 可以把两个 CompletableFuture 组合在一起。如果要是想组合任意数量的 CompletableFuture应该怎么做
可以使用 allOf 和 anyOf 组合任意多个 CompletableFuture。这两个函数都是静态函数参数是变长的 CompletableFuture 的集合。
allOf 和 anyOf 的区别前者是「与」后者是「或」。
allOf()
allOf 的返回值是 CompletableFutureVoid 类型这是因为每个传入的 CompletableFuture 的返回值都可能不同所以组合的结果是 无法用某种类型来表示的索性返回 Void 类型。那么如何获取每个 CompletableFuture 的执行结果呢
例子 并行地下载 N 个资源待下载完成之后并资源额外处理。
Test
public void allOf() throws Exception {// URL 列表集合ListString webLinks Arrays.asList(https://www.baidu.com/, https://www.bing.com/, https://www.so.com/);// 并行下载多个网页ListCompletableFutureString contentFutureList webLinks.stream().map(webLink - downloadWebLink(webLink)).collect(Collectors.toList());// 通过allof,等待所有网页下载完毕收集返回结果CompletableFutureVoid allFutures CompletableFuture.allOf(contentFutureList.toArray(new CompletableFuture[contentFutureList.size()]));// 附上回调函数获取结果集// 方式一CompletableFutureListString result allFutures.thenApply(v - contentFutureList.stream().map(o - {try {return o.get();} catch (InterruptedException e) {throw new RuntimeException(e);} catch (ExecutionException e) {throw new RuntimeException(e);}}).collect(Collectors.toList()));System.out.println(result.get());// 方式二result allFutures.thenApply(v - contentFutureList.stream().map(CompletableFutureString::join).collect(Collectors.toList()));System.out.println(result.get());
}private CompletableFutureString downloadWebLink(String webLink) {return CompletableFuture.supplyAsync(() - {// 模拟下载过程System.out.println(开始下载网页 webLink);return 这是一个网页内容;});}
这里有个关键问题因为 allof 没有返回值所以通过 theApply给 allFutures 附上一个回调函数。在回调函数里面以此调用么一个 Future 的 Get() 函数获取结果后存入 ListString 中。
anyOf()
anyOf 的含义是只要有任意一个 CompletableFuture 结束就可以做接下来的事情而无须像 allOf 那样等待所有的 CompletableFuture 结束。
但由于每个 CompletableFuture 的返回值类型可能不同意味着无法判断是什么类型所以 anyOf 的返回值是 CompletableFutureObject 类型。
Test
public void anyOf() throws Exception {CompletableFutureString future1 CompletableFuture.supplyAsync(() - {try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {throw new RuntimeException(e);}return future1 结果;});CompletableFutureString future2 CompletableFuture.supplyAsync(() - {try {TimeUnit.SECONDS.sleep(4);} catch (InterruptedException e) {throw new RuntimeException(e);}return future2 结果;});CompletableFutureString future3 CompletableFuture.supplyAsync(() - {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}return future3 结果;});CompletableFutureObject future CompletableFuture.anyOf(future1, future2, future3);System.out.println(future.get());
}
在该例子中因为 future1、future2、future3 的返回值都是 CompletableFutureString所以anyOf 的返回的 Object 一定也是 String 类型。
并且在 3 个 future 中future3 睡眠时间最短会最先执行完成 anyOfFuture.get() 获取的也就是 future3 的内容。future1、future2 的返回结果会被丢弃。
异常处理
在调用 supplyAsync() 任务中发生一个错误这时候没有任何 thenApply 会被调用并且 future 将以一个异常结束。如果在第一个 thenApply() 发生错误这时候第二个和第三个将不会被调用同样的future 将以异常结束。
exceptionally()
回调处理异常从原始Future中生成的错误恢复的机会。
Test
public void exceptionally() throws Exception {Integer age -1;CompletableFutureString future CompletableFuture.supplyAsync(() - {if (age 0) {throw new IllegalArgumentException(年龄不能为负数);}return 张三;}).exceptionally(ex - {System.out.println(ex.getMessage());return Unknown!;});System.out.println(future.get());
}
handle()
从异常恢复无论一个异常是否发生它都会被调用。
Test
public void handle() throws Exception {Integer age -1;CompletableFutureString future CompletableFuture.supplyAsync(() - {if (age 0) {throw new IllegalArgumentException(年龄不能为负数);}return 张三;}).handle((res, ex) - {if (ex ! null) {System.out.println(ex.getMessage());return Unknown!;}return res;});System.out.println(future.get());
}
如果异常发生 res 参数将是 null否则 ex 将是 null。
参考CompletableFuture原理与用法详解
