CompletableFuture超详解与实践

0.背景

一个接口可能需要调用 N 个其他服务的接口,这在项目开发中还是挺常见的。举个例子:用户请求获取订单信息,可能需要调用用户信息、商品详情、物流信息、商品推荐等接口,最后再汇总数据统一返回。

如果是串行(按顺序依次执行每个任务)执行的话,接口的响应速度会非常慢。考虑到这些接口之间有大部分都是 无前后顺序关联 的,可以 并行执行 ,就比如说调用获取商品详情的时候,可以同时调用获取物流信息。通过并行执行多个任务的方式,接口的响应速度会得到大幅优化。

serial-to-parallel

对于存在前后顺序关系的接口调用,可以进行编排,如下图所示。

img

在外卖商家端交易业务存在这样的特点:

1.服务端必须一次返回订单卡片所有内容:根据商家端和服务端的“增量同步协 议注 1”,服务端必须一次性返回订单的所有信息,包含订单主信息、商品、结算、配送、用户信息、骑手信息、餐损、退款、客服赔付(参照下面订单卡片截图)等,需要从下游三十多个服务中获取数据。在特定条件下,如第一次登录和长时间没登录的情况下,客户端会分页拉取多个订单,这样发起的远程调用会更多。

2.商家端和服务端交互频繁:商家对订单状态变化敏感,多种推拉机制保证每次变更能够触达商家,导致 App 和服务端的交互频繁,每次变更需要拉取订单最新的全部内容在外卖交易链路如此大的流量下,为了保证商家的用户体验,保证接口的高性能,并行从下游获取数据就成为必然

综上所述,在多模块大流量的场景下原先的串行接口的响应是行不通的,所以并行接口的响应是必然的

1.CompletableFuture使用与原理

1.1Future介绍

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

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

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

  • 取消任务;
  • 判断任务是否被取消;
  • 判断任务是否已经执行完成;
  • 获取任务执行结果。
// V 代表了Future执行的任务返回值的类型
public interface Future<V> {// 取消任务执行// 成功取消返回 true,否则返回 falseboolean cancel(boolean mayInterruptIfRunning);// 判断任务是否被取消boolean isCancelled();// 判断任务是否已经执行完成boolean isDone();// 获取任务执行结果V get() throws InterruptedException, ExecutionException;// 指定时间内没有返回计算结果就抛出 TimeOutException 异常V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutExceptio}
//举例
public class FutureTest {public static void main(String[] args) throws InterruptedException, ExecutionException {ExecutorService executorService = Executors.newFixedThreadPool(5);UserInfoService userInfoService = new UserInfoService();MedalService medalService = new MedalService();long userId = 666L;long startTime = System.currentTimeMillis();//调用用户服务获取用户基本信息FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() {@Overridepublic UserInfo call() throws Exception {return userInfoService.getUserInfo(userId);}});executorService.submit(userInfoFutureTask);Thread.sleep(300); //模拟主线程其它操作耗时FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() {@Overridepublic MedalInfo call() throws Exception {return medalService.getMedalInfo(userId);}});executorService.submit(medalInfoFutureTask);UserInfo userInfo = userInfoFutureTask.get();//获取个人信息结果MedalInfo medalInfo = medalInfoFutureTask.get();//获取勋章信息结果System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");}
}class UserInfoService {public UserInfo getUserInfo(Long userId) throws InterruptedException {Thread.sleep(300);//模拟调用耗时return new UserInfo("666", "捡田螺的小男孩", 27); //一般是查数据库,或者远程调用返回的}}class MedalService {public MedalInfo getMedalInfo(long userId) throws InterruptedException {Thread.sleep(500); //模拟调用耗时return new MedalInfo("666", "守护勋章");}
}

如果我们不使用Future进行并行异步调用,而是在主线程串行进行的话,耗时大约为300+500+300 = 1100 ms。可以发现,future+线程池异步配合,提高了程序的执行效率。

但是Future对于结果的获取,不是很友好,只能通过阻塞或者轮询的方式得到任务的结果。

  • Future.get() 就是阻塞调用,在线程获取结果之前get方法会一直阻塞
  • Future提供了一个isDone方法,可以在程序中轮询这个方法查询执行结果。

阻塞的方式和异步编程的设计理念相违背,而轮询的方式会耗费无谓的CPU资源。因此,JDK8设计出CompletableFuture。CompletableFuture提供了一种观察者模式类似的机制,可以让任务执行完成后通知监听的一方。

1.2CompletableFuture的背景和定义

1.2.1CompletableFuture解决的问题

CompletableFuture 是由 Java 8 引入的,在 Java8 之前我们一般通过 Future 实现 异步。

● Future 用于表示异步计算的结果,只能通过阻塞或者轮询的方式获取结果, 而且不支持设置回调方法,Java 8 之前若要设置回调一般会使用 guava 的ListenableFuture,回调的引入又会导致臭名昭著的回调地狱(下面的例子会通过 ListenableFuture 的使用来具体进行展示)。

● CompletableFuture 对 Future 进行了扩展,可以通过设置回调的方式处理计

算结果,同时也支持组合操作,支持进一步的编排,同时一定程度解决了回调地狱的问题。

下面将举例来说明,我们通过 ListenableFuture、CompletableFuture 来实现异步 的差异。假设有三个操作 step1、step2、step3 存在依赖关系,其中 step3 的执行依赖 step1 和 step2 的结果。

Future(ListenableFuture) 的实现(回调地狱)如下:

 ExecutorService executor = Executors.newFixedThreadPool(5);ListeningExecutorService guavaExecutor = MoreExecutors.listeningDecorator(executor);ListenableFuture<String> future1 = guavaExecutor.submit(() -> {//step 1System.out.println(“执行 step 1);return“step1 result”;});ListenableFuture<String> future2 = guavaExecutor.submit(() -> {//step 2System.out.println(“执行 step 2);return“step2 result”;});ListenableFuture<List<String>> future1And2 = Futures.allAsList(future1,future2);
Futures.addCallback(future1And2,new FutureCallback<List<String>>(){@Overridepublic void onSuccess (List < String > result) {System.out.println(result);ListenableFuture<String> future3 = guavaExecutor.submit(() -> {System.out.println(“执行 step 3);return“step3 result”;});Futures.addCallback(future3, new FutureCallback<String>() {@Overridepublic void onSuccess(String result) {System.out.println(result);}@Overridepublic void onFailure(Throwable t) {}}, guavaExecutor);}@Overridepublic void onFailure (Throwable t){}},guavaExecutor);

CompletableFuture 的实现如下:

ExecutorService executor = Executors.newFixedThreadPool(5);CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {System.out.println(“执行 step 1);}return“step1 result”;
}, executor);
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {System.out.println(“执行 step 2);return“step2 result”;
});
cf1.thenCombine(cf2, (result1, result2) -> {System.out.println(result1 +,+ result2);System.out.println(“执行 step 3);return“step3 result”;
}).thenAccept(result3 -> System.out.println(result3));

显然,CompletableFuture 的实现更为简洁,可读性更好。

1.2.1 CompletableFuture 的定义

img

CompletableFuture 实现了两个接口(如上图所示):Future、CompletionStage。 Future 表示异步计算的结果,CompletionStage 用于表示异步执行过程中的一个步 骤(Stage),这个步骤可能是由另外一个 CompletionStage 触发的,随着当前步骤的完成,也可能会触发其他一系列 CompletionStage 的执行。从而我们可以根据实际业务对这些步骤进行多样化的编排组合,CompletionStage 接口正是定义了这样的能力,我们可以通过其提供的 thenAppythenCompose 等函数式编程方法来组合编排这些步骤。

1.3CompletableFuture 的使用

下面我们通过一个例子来讲解 CompletableFuture 如何使用,使用 CompletableFuture 也是构建依赖树的过程。一个 CompletableFuture 的完成会触发另外一系列依赖它的 CompletableFuture 的执行

在这里插入图片描述

如上图所示,这里描绘的是一个业务接口的流程,其中包括 CF1\CF2\CF3\CF4\CF5 共 5 个步骤,并描绘了这些步骤之间的依赖关系,每个步骤可以是一次 RPC 调用、一次数据库操作或者是一次本地方法调用等,在使用 CompletableFuture 进行 异步化编程时,图中的每个步骤都会产生一个 CompletableFuture 对象,最终结果

也会用一个 CompletableFuture 来进行表示。根据 CompletableFuture 依赖数量,可以分为以下几类:零依赖、一元依赖、二元依赖和多元依赖

1.3.1 零依赖:CompletableFuture 的创建

我们先看下如何不依赖其他 CompletableFuture 来创建新的 CompletableFuture:

在这里插入图片描述

如上图红色链路所示,接口接收到请求后,首先发起两个异步调用 CF1、CF2,主要有三种方式:

  ExecutorService executor = Executors.newFixedThreadPool(5);//1、使用 runAsync 或 supplyAsync 发起异步调用CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {return "result1";}, executor);//2、CompletableFuture.completedFuture() 直接创建一个已完成状态的CompletableFutureCompletableFuture<String> cf2 = CompletableFuture.completedFuture("result2");//3、先初始化一个未完成的 CompletableFuture,然后通过 complete()、completeExceptionally(),完成该 CompletableFutureCompletableFuture<String> cf = new CompletableFuture<>();cf.complete("success");

第三种方式的一个典型使用场景,就是将回调方法转为 CompletableFuture,然后再依赖 CompletableFure 的能力进行调用编排,示例如下:

@FunctionalInterface
public interface ThriftAsyncCall {void invoke() throws TException;
}/*** 该方法为美团内部 rpc 注册监听的封装,可以作为其他实现的参照* OctoThriftCallback 为 thrift 回调方法* ThriftAsyncCall 为自定义函数,用来表示一次 thrift 调用(定义如上)*/static <T> CompletableFuture<T> toCompletableFuture(finalOctoThriftCallback<?, T> callback, ThriftAsyncCall thriftCall) {// 新建一个未完成的 CompletableFutureCompletableFuture<T> resultFuture = new CompletableFuture<>();// 监听回调的完成,并且与 CompletableFuture 同步状态callback.addObserver(new OctoObserver<T>() {@Overridepublic void onSuccess(T t) {resultFuture.complete(t);}@Overridepublic void onFailure(Throwable throwable) {resultFuture.completeExceptionally(throwable);}});if (thriftCall != null) {try {thriftCall.invoke();} catch (TException e) {resultFuture.completeExceptionally(e);}}return resultFuture;}
1.3.2 一元依赖:依赖一个 CF

在这里插入图片描述

如上图红色链路所示,CF3,CF5 分别依赖于 CF1 和 CF2,这种对于单个 CompletableFuture 的依赖可以通过 thenApplythenAcceptthenCompose 等方法来实现,代码如下所示:

CompletableFuture<String> cf3 = cf1.thenApply(result1 -> {//result1 为 CF1 的结果//......return“result3”;
});
CompletableFuture<String> cf5 = cf2.thenApply(result2 -> {//result2 为 CF2 的结果//......return“result5”;
});
1.3.3 二元依赖:依赖两个 CF

在这里插入图片描述

如上图红色链路所示,CF4 同时依赖于两个 CF1 和 CF2,这种二元依赖可以通过thenCombine 等回调来实现,如下代码所示:

CompletableFuture<String> cf4 = cf1.thenCombine(cf2, (result1, result2) -> {//result1 和 result2 分别为 cf1 和 cf2 的结果return“result4”;
});
1.3.4 多元依赖:依赖多个 CF

在这里插入图片描述

如上图红色链路所示,整个流程的结束依赖于三个步骤 CF3、CF4、CF5,这种多元依赖可以通过 allOf 或 anyOf 方法来实现,区别是当需要多个依赖全部完成时使用 allOf,当多个依赖中的任意一个完成即可时使用 anyOf,如下代码所示:

CompletableFuture<Void> cf6 = CompletableFuture.allOf(cf3, cf4, cf5);
CompletableFuture<String> result = cf6.thenApply(v -> {// 这里的 join 并不会阻塞,因为传给 thenApply 的函数是在 CF3、CF4、CF5 全部完成时,才会执行 。result3 = cf3.join();result4 = cf4.join();result5 = cf5.join();// 根据 result3、result4、result5 组装最终 result;return“result”;
});

1.4CompletableFuture 原理

CompletableFuture 中包含两个字段:resultstack。result 用于存储当前 CF的结果,stack(Completion)表示当前 CF 完成后需要触发的依赖动作(Dependency Actions),去触发依赖它的 CF 的计算,依赖动作可以有多个(表示有多个依赖它的 CF),以栈(Treiber stack)的形式存储,stack 表示栈顶元素

在这里插入图片描述

这种方式类似“观察者模式”,依赖动作(Dependency Action)都封装在一个单独Completion 子类中。下面是 Completion 类关系结构图。CompletableFuture 中的每个方法都对应了图中的一个 Completion 的子类,Completion 本身是观察者的基类。

1.UniCompletion 继承了 Completion,是一元依赖的基类,例如 thenApply的实现类 UniApply 就继承自 UniCompletion。

2.BiCompletion 继承了 UniCompletion,是二元依赖的基类,同时也是多元依赖的基类。例如 thenCombine 的实现类 BiRelay 就继承自 BiCompletion。

在这里插入图片描述

1.4.1 CompletableFuture 的设计思想

按照类似“观察者模式”的设计思想,原理分析可以从“观察者”和“被观察者”两个方面着手。由于回调种类多,但结构差异不大,所以这里单以一元依赖中的thenApply 为例,不再枚举全部回调类型。如下图所示:

在这里插入图片描述

1.4.1.1 被观察者
  1. 每 个 CompletableFuture 都 可 以 被 看 作 一 个 被 观 察 者, 其 内 部 有 一 个
    Completion 类型的链表成员变量 stack,用来存储注册到其中的所有观察者。当被观察者执行完成后会弹栈 stack 属性,依次通知注册到其中的观察者。上面例子中步骤 fn2 就是作为观察者被封装在 UniApply 中。
  2. 被观察者 CF 中的 result 属性,用来存储返回结果数据。这里可能是一次RPC 调用的返回值,也可能是任意对象,在上面的例子中对应步骤 fn1 的执行结果
1.4.1.2 观察者

CompletableFuture 支持很多回调方法,例如 thenAccept、thenApply、exceptionally 等,这些方法接收一个函数类型的参数 f,生成一个 Completion 类型的对象(即观察者),并将入参函数 f 赋值给 Completion 的成员变量 fn,然后检查当前CF 是否已处于完成状态(即 result != null),如果已完成直接触发 fn,否则将观察者
Completion 加入到 CF 的观察者链 stack 中,再次尝试触发,如果被观察者未执行完则其执行完毕之后通知触发。

  1. 观察者中的 dep 属性:指向其对应的 CompletableFuture,在上面的例子中dep 指向 CF2。
  2. 观察者中的 src 属性:指向其依赖的 CompletableFuture,在上面的例子中src 指向 CF1。
  3. 观察者 Completion 中的 fn 属性:用来存储具体的等待被回调的函数。这里需要注意的是不同的回调方法(thenAccept、thenApply、exceptionally 等)接收的函数类型也不同,即 fn 的类型有很多种,在上面的例子中 fn 指向 fn2。
1.4.2 整体流程
1.4.2.1 一元依赖

这里仍然以 thenApply 为例来说明一元依赖的流程:

  1. 将观察者 Completion 注册到 CF1,此时 CF1 将 Completion 压栈。
  2. 当 CF1 的操作运行完成时,会将结果赋值给 CF1 中的 result 属性。
  3. 依次弹栈,通知观察者尝试运行。

在这里插入图片描述

初步流程设计如上图所示,这里有几个关于注册与通知的并发问题,大家可以思考下:

Q1:在观察者注册之前,如果 CF 已经执行完成,并且已经发出通知,那么这时观察者由于错过了通知是不是将永远不会被触发呢 ?

A1:不会。在注册时检查依赖的 CF 是否已经完成。如果未完成(即 result == null)则将观察者入栈,如果已完成(result != null)则直接触发观察者操作。
Q2:在”入栈“前会有”result == null“的判断,这两个操作为非原子操作,CompletableFufure 的实现也没有对两个操作进行加锁,完成时间在这两个操作之间,观察者仍然得不到通知,是不是仍然无法触发?

在这里插入图片描述

A2:不会。入栈之后再次检查 CF 是否完成,如果完成则触发。
Q3:当依赖多个 CF 时,观察者会被压入所有依赖的 CF 的栈中,每个 CF 完成的时候都会进行,那么会不会导致一个操作被多次执行呢 ?如下图所示,即当 CF1、CF2 同时完成时,如何避免 CF3 被多次触发。

在这里插入图片描述

A3:CompletableFuture 的实现是这样解决该问题的:观察者在执行之前会先通过 CAS 操作设置一个状态位,将 status 由 0 改为 1。如果观察者已经执行过了,那么 CAS 操作将会失败,取消执行

1.4.2.2thenApply源码分析
 public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) {return uniApplyStage(null, fn);}private <V> CompletableFuture<V> uniApplyStage(Executor e, Function<? super T,? extends V> f) {if (f == null) throw new NullPointerException();//1.创建一个新的CompletableFuture对象CompletableFuture<V> d =  new CompletableFuture<V>();if (e != null || !d.uniApply(this, f, null)) {//2.构建UniApplye代表线程池 d 代表新的CompletableFuture this 代表当前f 代表方法 这个时候 UniApply 内部的所有的引用都处于为null的状态UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);//3. c其实就是Completion对象,被push到栈中push(c);//4. 尝试执行cc.tryFire(SYNC);}// 5. 这个d会一直返回到调用thenApply的地方,后续的链式调用会作用在这个d上面return d;}@SuppressWarnings("serial")static final class UniApply<T,V> extends UniCompletion<T,V> {Function<? super T,? extends V> fn;UniApply(Executor executor, CompletableFuture<V> dep,CompletableFuture<T> src,Function<? super T,? extends V> fn) {//2.1 向上执行super(executor, dep, src); this.fn = fn;}}abstract static class UniCompletion<T,V> extends Completion {Executor executor;                 // executor to use (null if none)CompletableFuture<V> dep;          // the dependent to completeCompletableFuture<T> src;          // source for actionUniCompletion(Executor executor, CompletableFuture<V> dep,CompletableFuture<T> src) {//2.2 dep就是新创建的d  src就是当前的thisthis.executor = executor; this.dep = dep; this.src = src;}}

关于执行第2步的时候,构建的对象如下图, src和dep都是空的CompletableFuture,next为Null,这里我们会发现所有的都是继承Completion对象,最终所有都是构建都可以理解为Completion对象;

关于执行第3步的时候,构建的UniApply对象的内容完成压栈的操作,将CompletableFuture的stack属性指向Completion对象;

接下来看第4步操作,尝试执行Completion;

 @SuppressWarnings("serial")static final class UniApply<T,V> extends UniCompletion<T,V> {Function<? super T,? extends V> fn;UniApply(Executor executor, CompletableFuture<V> dep,CompletableFuture<T> src,Function<? super T,? extends V> fn) {super(executor, dep, src); this.fn = fn;}final CompletableFuture<V> tryFire(int mode) {//4.1 d新创建的 a(也是c中的src) 就是原来的CompletableFuture<V> d; CompletableFuture<T> a;//4.2 如果uniApply执行成功,则会进到下面的postFire调用否则返回null 如果返回null,就要等待以后的主动complete来再次触发if ((d = dep) == null ||!d.uniApply(a = src, fn, mode > 0 ? null : this))return null;//4.5 tryFire成功后,会把以下几个属性设为null,表面此Completion已经完成任务,变成dead状态dep = null; src = null; fn = null;//4.6 出栈return d.postFire(a, mode);}}final <S> boolean uniApply(CompletableFuture<S> a,Function<? super S,? extends T> f,UniApply<S,T> c) {Object r; Throwable x;//4.3 如果a(也是c中的src)没有准备完成,那result是空,这里就会直接返回falseif (a == null || (r = a.result) == null || f == null)return false;tryComplete: if (result == null) {if (r instanceof AltResult) {if ((x = ((AltResult)r).ex) != null) {completeThrowable(x, r);break tryComplete;}r = null;}try {if (c != null && !c.claim())return false;@SuppressWarnings("unchecked") S s = (S) r;//4.4 如果r不为空,则会作为f的输入参数,f的输出则成为当前CompletableFuture的完成值completeValue(f.apply(s));} catch (Throwable ex) {completeThrowable(ex);}}return true;}

第5步返回d, 这个d会返回到调用thenApply的地方,后续的链式调用会作用在这个d上面,接下来我们可以看到base对象就是我们构建好的第一个链;

这里我们可以猜测后续的执行thenApply的方法,也就是执行完成test1的第二行代码,生成的结构如下图

在这里插入图片描述

当我们的代码执行到test1的第3行的时候,也就是complete方法,该方法也就是为了解决我们执行tryFire执行失败后动作,源码如下:

public boolean complete(T value) {boolean triggered = completeValue(value);postComplete();return triggered;}final void postComplete() {//1. this表示当前的CompletableFuture, 也就是我们baseCompletableFuture<?> f = this; Completion h;//2. 判断stack是否为空  或者如果f的栈为空且不是this则重置while ((h = f.stack) != null ||(f != this && (h = (f = this).stack) != null)) {CompletableFuture<?> d; Completion t;//3. CAS出栈if (f.casStack(h, t = h.next)) {if (t != null) { //4.出栈的h不是最后一个元素,最后一个元素直接执行7即可if (f != this) {//5. 如果f不是this,将刚出栈的h, 入this的栈顶//我猜测这个地方大家会有迷惑pushStack(h);continue;}h.next = null;    //6. detach}f = (d = h.tryFire(NESTED)) == null ? this : d;  //7.调用tryFire}}}
1.4.2.3 二元依赖

我们以 thenCombine 为例来说明二元依赖:

在这里插入图片描述

thenCombine 操作表示依赖两个 CompletableFuture。其观察者实现类为 BiApply,
如上图所示,BiApply 通过 src 和 snd 两个属性关联被依赖的两个 CF,fn 属性的类型为 BiFunction。与单个依赖不同的是,在依赖的 CF 未完成的情况下,thenCombine 会尝试将 BiApply 压入这两个被依赖的 CF 的栈中,每个被依赖的 CF 完成时都会尝试触发观察者 BiApply,BiApply 会检查两个依赖是否都完成,如果完成则开始执行。这里为了解决重复触发的问题,同样用的是上一章节提到的 CAS 操作,执行时会先通过 CAS 设置状态位,避免重复触发。

1.4.2.4 多元依赖

依赖多个 CompletableFuture 的回调方法包括 allOf、anyOf,区别在于 allOf观察者实现类为 BiRelay,需要所有被依赖的 CF 完成后才会执行回调;而 anyOf 观 察者实现类为 OrRelay,任意一个被依赖的 CF 完成后就会触发。二者的实现方式都是将多个被依赖的 CF 构建成一棵平衡二叉树,执行结果层层通知,直到根节点,触发回调监听

在这里插入图片描述

1.4.3 小结

本章节为 CompletableFuture 实现原理的科普,旨在尝试不粘贴源码,而通过结构图、流程图以及搭配文字描述把 CompletableFuture 的实现原理讲述清楚。把晦 涩的源码翻译为“整体流程”章节的流程图,并且将并发处理的逻辑融入,便于大家理解

2.实践总结

2.1线程阻塞问题:

2.1.1 代码执行在哪个线程上?

要合理治理线程资源,最基本的前提条件就是要在写代码时,清楚地知道每一行代码都将执行在哪个线程上。下面我们看一下 CompletableFuture 的执行线程情况。CompletableFuture 实现了 CompletionStage 接口,通过丰富的回调方法,支持各种组合操作,每种组合场景都有同步和异步两种方法。
同步方法(即不带 Async 后缀的方法)有两种情况。

● 如果注册时被依赖的操作已经执行完成,则直接由当前线程执行。

● 如果注册时被依赖的操作还未执行完,则由回调线程执行

异步方法(即带 Async 后缀的方法):可以选择是否传递线程池参数 Executor 运行在指定线程池中;当不传递 Executor 时,会使用 ForkJoinPool 中的共用线程池CommonPool(CommonPool 的大小是 CPU 核数 -1,如果是 IO 密集的应用,线程数可能成为瓶颈)。

ExecutorService threadPool1 = new ThreadPoolExecutor(10, 10, 0L, 
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100));
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {System.out.println(“supplyAsync 执行线程:” + Thread.currentThread().
getName());// 业务操作return“”;
}, threadPool1);
// 此时,如果 future1 中的业务操作已经执行完毕并返回,则该 thenApply 直接由当前
main 线程执行;否则,将会由执行以上业务操作的 threadPool1 中的线程执行。
future1.thenApply(value -> {System.out.println(“thenApply 执行线程:” + Thread.currentThread().
getName());return value +1;
});
// 使用 ForkJoinPool 中的共用线程池 CommonPool
future1.thenApplyAsync(value -> {
//do somethingreturn value +1;});
// 使用指定线程池
future1.thenApplyAsync(value -> {
//do somethingreturn value +1;
}, threadPool1);
2.1.2 异步回调要传线程池

前面提到,异步回调方法可以选择是否传递线程池参数 Executor,这里我们建议强制传线程池,且根据实际情况做线程池隔离。当不传递线程池时,会使用 ForkJoinPool 中的公共线程池 CommonPool,这里所有调用将共用该线程池,核心线程数 = 处理器数量 -1(单核核心线程数为 1),所有异步回调都会共用该 CommonPool,核心与非核心业务都竞争同一个池中的线程,很容易成为系统瓶颈。手动传递线程池参数可以更方便的调节参数,并且可以给不同的业务分配不同的线程池,以求资源隔离,减少不同业务之间的相互干扰。

2.1.3 线程池循环引用会导致死锁
public Object doGet() {ExecutorService threadPool1 = new ThreadPoolExecutor(10, 10, 0L, 
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100));CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {//do sthreturn CompletableFuture.supplyAsync(() -> {System.out.println(“child”);return“child”;}, threadPool1).join();// 子任务}, threadPool1);return cf1.join();
}

如上代码块所示,doGet 方法第三行通过 supplyAsync 向 threadPool1 请求线程,并且内部子任务又向 threadPool1 请求线程。threadPool1 大小为 10,当同一时刻有 10 个请求到达,则 threadPool1 被打满,子任务请求线程时进入阻塞队列排队,但是父任务的完成又依赖于子任务,这时由于子任务得不到线程,父任务无法完成。

主线程执行 cf1.join() 进入阻塞状态,并且永远无法恢复。为了修复该问题,需要将父任务与子任务做线程池隔离,两个任务请求不同的线程池,避免循环依赖导致的阻塞

2.1.4 异步 RPC 调用注意不要阻塞 IO 线程池

服务异步化后很多步骤都会依赖于异步 RPC 调用的结果,这时需要特别注意一点,如果是使用基于 NIO(比如 Netty)的异步 RPC,则返回结果是由 IO 线程负责设置的,即回调方法由 IO 线程触发,CompletableFuture 同步回调(如 thenApply、thenAccept 等无 Async 后缀的方法)如果依赖的异步 RPC 调用的返回结果,那么
这些同步回调将运行在 IO 线程上,而整个服务只有一个 IO 线程池,这时需要保证同步回调中不能有阻塞等耗时过长的逻辑,否则在这些逻辑执行完成前,IO 线程将一直被占用,影响整个服务的响应。

2.2 其他

2.2.1 异常处理

由于异步执行的任务在其他线程上执行,而异常信息存储在线程栈中,因此当前线程除非阻塞等待返回结果,否则无法通过 try\catch 捕获异常。CompletableFuture提供了异常捕获回调 exceptionally,相当于同步调用中的try\catch。使用方法如下所示:

@Autowired
private WmOrderAdditionInfoThriftService 
wmOrderAdditionInfoThriftService;// 内部接口
public CompletableFuture<Integer> getCancelTypeAsync(long orderId) {CompletableFuture<WmOrderOpRemarkResult> 
remarkResultFuture = wmOrderAdditionInfoThriftService.
findOrderCancelledRemarkByOrderIdAsync(orderId);// 业务方法,内部会发起异步rpc 调用return remarkResultFuture.exceptionally(err -> {// 通过 exceptionally 捕获异常,打印日志并返回默认值log.error(WmOrderRemarkService.getCancelTypeAsync Exception 
orderId={}, orderId, err);
return 0;});
}

3.异步化收益

通过异步化改造,API 系统的性能得到明显提升,与改造前对比的收益

如下:

● 核心接口吞吐量大幅提升,其中订单轮询接口改造前 TP99 为 754ms,改造后降为 408ms。

● 服务器数量减少 1/3。
参考文献:
《2022年美团技术年货-合辑》

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/606212.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

前端如何结合数据库查看本地头像文件,配置静态资源

由于oss过期了&#xff0c;项目中又需要一个上传头像功能&#xff0c;于是研究了一下上传文件至本地&#xff0c;用java将文件存储到本地的代码好写&#xff0c;但是如何利用前端Vue把我难住了&#xff0c;因为之前存在OSS服务器直接查看就可以了&#xff0c;于是记录本文。 后…

数据结构-测试4

一、判断题 1.队列结构的顺序存储会产生假溢出现象。 &#xff08;T&#xff09; 2.度为二的树就是二叉树。(F) 二叉树的度可以小于等于2 3. 栈是插入和删除只能在一端进行的线性表&#xff1b;队列是插入在一端进行&#xff0c;删除在另一端进行的线性表。&#xff08;T&…

C++模板——(2)函数模板的声明和使用

归纳编程学习的感悟&#xff0c; 记录奋斗路上的点滴&#xff0c; 希望能帮到一样刻苦的你&#xff01; 如有不足欢迎指正&#xff01; 共同学习交流&#xff01; &#x1f30e;欢迎各位→点赞 &#x1f44d; 收藏⭐ 留言​&#x1f4dd; 勤奋&#xff0c;机会&#xff0c;&am…

Linux文件系统与日志管理

目录 一、Linux文件系统 1、inode 与 block 详解 1.1 inode 和 block 概述 1.2 inode表的内容 1.3 查看文件的inode号码 1.4 模拟innode号耗尽故障处理 2、访问文件的流程 3、文件恢复 3.1 恢复误删除的ext3格式文件 3.2 恢复误删除的 xfs 格式文件 二、Linux日志…

《BackTrader量化交易图解》第8章:plot 绘制金融图

文章目录 8. plot 绘制金融图8.1 金融分析曲线8.2 多曲线金融指标8.3 Observers 观测子模块8.4 plot 绘图函数的常用参数8.5 买卖点符号和色彩风格 8. plot 绘制金融图 8.1 金融分析曲线 BackTrader内置的plot绘图函数&#xff0c; 通过style参数变量控制不同风格的K线图&…

因为热Key和大Key,Redis终于被压崩了

大家好&#xff0c;我是洋子&#xff0c;今天分享一下在做压测时遇到的很有意思的性能问题以及对应的排查解决方案。这个性能问题的现象为&#xff0c;Redis线上实例不可用&#xff0c;Redis读写均超时 性能问题排查过程 先来看一下问题代码&#xff08;Go语言实现&#xff0…

【CMake】1. VSCode 开发环境安装与运行

CMake 示例工程代码 https://github.com/LABELNET/cmake-simple 插件 使用 VSCode 开发C项目&#xff0c;安装 CMake 插件 CMakeCMake ToolsCMake Language Support &#xff08;建议&#xff0c;语法提示) 1. 配置 CMake Language Support , Windows 配置 donet 环境 这…

RT-Thread: 基于STM32CubeMX配置驱STM32驱动的USB虚拟串口调试

关键词&#xff1a;USB 虚拟串口 USB虚拟串口&#xff0c;RT-Thread Studio&#xff0c;STM32 说明&#xff1a; 1&#xff1a;文档记录 STM32F103系列基于 RT-Thread 系统的 USB虚拟串口的开启及数据收发应用流程介绍。 2&#xff1a;本文以STM32F103C8T6型号做测试&#x…

计算机基础面试题 |16.精选计算机基础面试题

&#x1f90d; 前端开发工程师&#xff08;主业&#xff09;、技术博主&#xff08;副业&#xff09;、已过CET6 &#x1f368; 阿珊和她的猫_CSDN个人主页 &#x1f560; 牛客高级专题作者、在牛客打造高质量专栏《前端面试必备》 &#x1f35a; 蓝桥云课签约作者、已在蓝桥云…

Win10电脑关闭OneDrive自动同步的方法

在Win10电脑操作过程中&#xff0c;用户想要关闭OneDrive的自动同步功能&#xff0c;但不知道具体要怎么操作&#xff1f;首先用户需要打开OneDrive&#xff0c;然后点击关闭默认情况下将文档保存到OneDrive选项保存&#xff0c;最后关闭在这台电脑上同步设置保存就好了。接下来…

TypeScript 从入门到进阶之基础篇(九) Class类篇

系列文章目录 TypeScript 从入门到进阶系列 TypeScript 从入门到进阶之基础篇(一) ts基础类型篇TypeScript 从入门到进阶之基础篇(二) ts进阶类型篇TypeScript 从入门到进阶之基础篇(三) 元组类型篇TypeScript 从入门到进阶之基础篇(四) symbol类型篇TypeScript 从入门到进阶…

OCP NVME SSD规范解读-5.命令超时限制-2

Sanitize清除的数据很彻底&#xff0c;对FTL映射表、User Data(包括已经写入NAND和仍在cache里的)、Meta Data、安全密匙、CMB中SQ/CQ相关信息、可能含有用户数据的log等等会全部清除。不过&#xff0c;sanitize操作不会改变RPMB、boot分区、不包含用户数据的cache等内容。 RP…

总420+,专业120+南京大学851信号与系统电子信息考研经验通信,电子信息

今年考研数学130&#xff0c;专业课120&#xff0c;总分420顺利被南京大学电通录取&#xff0c;梦圆南大&#xff0c;这一年的复习有过迷茫&#xff0c;有过犹豫&#xff0c;最后都坚持过来了&#xff0c;总结一下自己的复习经验&#xff0c;希望对大家有所帮助。数学 5-8月数…

HarmonyOS应用开发之ArkTS语言学习记录

1、ArkTS介绍 ArkTS是鸿蒙生态的应用开发语言。它在保持TypeScript&#xff08;简称TS&#xff09;基本语法风格的基础上&#xff0c;对TS的动态类型特性施加更严格的约束&#xff0c;引入静态类型。同时&#xff0c;提供了声明式UI、状态管理等相应的能力&#xff0c;让开发者…

76.乐理基础-打拍子-二连音、四连音

内容来源于&#xff1a;三分钟音乐社 上一个内容&#xff1a;八三、八六拍的三角形打法-CSDN博客 这里要先理解了三连音的知识。 关于多少连音的总方针&#xff0c;其实就是两句话&#xff0c;如下图中的内容&#xff1a;二连音与四连音实际上就是下图中第二句话里的第一部分…

mmdetection训练自己的数据集

mmdetection训练自己的数据集 这里写目录标题 mmdetection训练自己的数据集一&#xff1a; 环境搭建二&#xff1a;数据集格式转换(yolo转coco格式)yolo数据集格式coco数据集格式yolo转coco数据集格式yolo转coco数据集格式的代码 三&#xff1a; 训练dataset数据文件配置config…

高防ip适合防御网站和游戏类的攻击吗?

​  作为站长&#xff0c;要学会并承受得住网站外来攻击的压力&#xff0c;尤其是所属为 DDoS 攻击高发行业的网站类业务及游戏行业&#xff0c;是很容易被竞争对手或者一些伪黑客爱好者盯上的。 加上&#xff0c;有些站长并没有提前了解&#xff0c;就盲目进军了这两个行业&…

C语言算法(二分查找、文件读写)

二分查找 前提条件&#xff1a;数据有序&#xff0c;随机访问 #include <stdio.h>int binary_search(int arr[],int n,int key);int main(void) {}int search(int arr[],int left,int right,int key) {//边界条件if(left > right) return -1;//int mid (left righ…

SAP 物料读取基本数据文本与检验文本READ_TEXT

1. 读取基本数据文本 使用函数 READ_TEXT 2. 读取检验文本

聚道云软件连接器助力某软件科技有限公司实现人力资源信息自动化

客户介绍&#xff1a; 某软件科技有限公司是一家集软件研发、销售、服务于一体的综合性软件企业。公司业务遍布全球多个国家和地区&#xff0c;拥有众多员工。 添加图片注释&#xff0c;不超过 140 字&#xff08;可选&#xff09; 客户痛点&#xff1a; 部门及人员信息分散…