消息队列面试解析系列之异步编程模式

更多精彩在公众号：JavaEdge

0 前言

线程就是为了能自动分配CPU时间片而生。异步模式设计可显著减少线程等待，在高吞吐量场景中，极大提升系统整体性能，降低时延。因此，像MQ这种需要超高吞吐量和超低时延中间件系统，其核心流程大量采用异步。

异步的本质是为了不占用过多的线程对象。比如一个响应时间是1秒的http1.1请求，并且不考虑http pipeline：

• 同步模式下，一个请求在未返回前，需要独占一个线程和一个httpconnection
• 异步模式下，一个请求在未返回前，只需要独占一个httpconnection,那个线程在提交完io任务后就回到线程池了

即QPS 5000时，同步需5000个connection和5000个线程，而异步可以省下5000个线程的内存以及操作系统对这些线程的管理能耗。

1 案例

某转账微服务Transfer有如下参数

• 转出账户
• 转入账户
• 转账金额

调用另外一个微服务Add(account, amount)，给账户account增加金额amount，当amount为负值时，就是扣减相应金额。现在要从账户A转账100到账户B：

1. 先从A的账户减100元
2. 再给B的账户加100元
3. 转账完成
   

2 同步性能瓶颈

假设Add平均响应时延60ms，Transfer平均响应时延就是120ms。Transfer每处理一个请求耗时120ms，这过程要独占1个线程。每个线程每s最多可处理约10个请求。假设服务器同时打开线程数量上限为10,000，可计算出这台服务器每s可处理请求上限： 10,000 （个线程）* 10（次请求每秒） = 100,000 次每秒。

若请求速度超过该值，请求就不能被马上处理，只能阻塞或排队，这时Transfer服务响应时延由120ms延长到：排队等待时延 + 处理时延(120ms)。即大量请求时，微服务平均响应时延变长！这就到了服务器极限吗？远没有！若监测服务器指标，会发现无论CPU、内存or网卡流量、磁盘I/O都闲的很，那Transfer服务那10,000个线程在作甚？绝大部分线程都在等待Add服务返回结果！所以采用同步，整个服务器的所有线程大部分时间都没在工作，而是在等待！若能减少或避免这种无意义等待，就能大幅提升服务吞吐量，提升性能。

3 异步方案

TransferAsync只是比Transfer多个参数，一个回调方法OnComplete（Java可传个回调类的实例来实现）：
请帮我执行转账，当转账完成后，请调用OnComplete。调用TransferAsync的线程无需等待转账完成，即可立即返回。待转账结束，TransferService自然会调用OnComplete()方法执行转账后续工作。

异步实现相比同步实现，先要定义如下回调方法：

• OnDebit()：扣减账户accountFrom完成后调用
• OnAllDone()：转入to账户完成后调用

异步实现的语义：

1. 异步从from账户扣减钱数，然后调用OnDebit
2. 在OnDebit中，异步将减去的钱数加到to账户，然后执行OnAllDone
3. 在OnAllDone中调用OnComplete
   

异步的时序流程和同步实现完全一样，只是线程模型由同步调用改为异步和回调。

性能分析

时序和同步实现一样，在少量请求场景下，平均响应时延一样是120ms。在高请求数量场景下，异步不再需线程等待执行结果，只需个位数量的线程，即可实现同步场景需要大量线程同样的吞吐量。

由于无线程数量限制，总体吞吐上限>>同步实现，且在服务器CPU、网络带宽资源达到极限前，响应时延不会随请求数量增加而显著升高，几乎可一直保持约120ms平均响应时延。

4 异步框架: CompletableFuture

Java开发常用异步框架：

• Java8内置的CompletableFuture

  CompletableFuture简单实用易理解

• ReactiveX的RxJava

  功能更强大

Java 8中新增的CompletableFuture几乎涵盖异步程序所需的大部分功能，易写出优雅且易维护的异步代码。

接下来用CompletableFuture改造转账服务。

微服务接口：

转账服务：

客户端使用CompletableFuture既可同步调用，也可异步：

调用异步方法获得返回的CompletableFuture对象后：

• 既可调用CompletableFuture#get，像调用同步方法样等待调用的方法执行结束并获得返回值
• 也能像异步回调，调用CompletableFuture#thenXXX，为CompletableFuture定义异步方法结束之后的后续操作
  比如上例，调用thenRun()方法，参数就是将转账完成打印在控台上这个操作，这样就可以实现在转账完成后，在控制台打印“转账完成！”了。

FAQ

异步实现中，若调用账户服务失败，如何将错误报告给客户端？在两次调用账户服务的Add方法时，若某一次调用失败了，该如何处理才能保证账户数据是平的？

• 调用账户失败，可以在异步callBack里执行通知客户端的逻辑
• 若是第一次失败，后面那步就不用执行了，所以转账失败；若第一次成功但是第二次失败，首先考虑重试，若转账服务幂等，可考虑一定次数的重试，若不能重试，考虑采用补偿，回滚第一次的转账操作。

异步实现中，回调方法OnComplete()在什么线程运行的？是否能控制回调方法的执行线程数？

CompletableFuture默认在ForkjoinPool commonpool里执行，也可指定一个Executor线程池执行，借鉴guava的ListenableFuture的时间，回调可以指定线程池执行，这样就能控制这个线程池的线程数。

异步实现中，回调方法 OnComplete()在执行OnAllDone()回调方法的那个线程，可通过一个异步线程池控制回调方法的线程数，如Spring中的async就是通过结合线程池来实现异步。

CompletableFuture回调底层还是forkjoin框架，forkjoin对I/O这种操作会阻塞线程且CompletableFuture默认线程数=cpu核数。在容器化场景下，CPU核数都不会很多，那使用CompletableFuture时，执行I/O操作会不会更早得无响应？因为个位数的线程很快就都被阻塞完了。

CompletableFuture不完全同于ForkJoin，可简单理解为：

• CompletableFuture.then() 等于 Fork
• CompletableFuture.get() 等于 Join

但并非所有场景下，CompletableFuture都要用get()结束，有时无需调用阻塞的get()方法。而且CompletableFuture默认使用 ForkJoinPool，但也支持给它提供一个自定义执行器。

异步是可以解决请求超时的问题，但是像文中举例这种转账操作，转出转入两个操作是前后依赖的没法并行，那么这种前后依赖的任务使用异步跟同步又有什么区别呢？另外，当10万请求过来之后，虽然用了异步可以瞬间返回，但是其实几万个请求对象在CompletableFuture内部线程池内部还是排队啊，所以最后来的请求还是要等很久才能被执行到。那么既然同步or异步都需要排队，异步究竟快在哪里了呢？

第一个问题，转入转出这两个操作不需要串行，是可以并行的。甚至执行顺序都没什么要求。我们唯一要保证的是这两个操作在一个事务中执行， “要么都成功，要么都失败”，就可以了。

你这个场景是在调用方（转账服务）异步，而服务提供方（账户服务）还是同步服务的情况下，才会出现。

你仔细看一下我们的异步设计，服务提供方提供的也是异步服务，那调用账户服务也是一瞬间就完成了，这样就不会出现你说的“几万个请求对象在CompletableFuture内部线程池内部还是排队”的情况了。

5 总结

异步思想就是，当要执行很耗时的操作时，不去等待操作结束，而是给该操作一个命令：“当ooo操作完成后，然后执行xxx”

使用异步编程，本身并不能加快程序本身的速度，但能减少或避免线程等待，只用很少线程就得到高吞吐。

异步性能虽好，切勿滥用，只有类似MQ这种业务逻辑简单且需超高吞吐量场景，或须长时等待资源，才考虑使用异步模型。
若业务逻辑复杂，在性能足够满足业务需求情况下，采用易于开发维护的同步模型更适合。