春招面试复习：消息队列（六）- 异步编程妙用

超、凢脫俗 2023-01-06 11:35 87阅读 0赞

# 0 异步的优势 #

太多的线程会造成频繁的cpu上下文切换，你可以想象一下，假设你的小公司只有8台电脑，你雇8个程序员一直不停的工作显然是效率最高的。考虑到程序员要休息不可能连轴转，雇佣24个人，每天三班倒，效率也还行。

但是，你要雇佣10000个人，他们还是只能用这8台电脑，大部分时间不都浪费在换人、交接工作上啦。

异步编程是通过分工的方式，是为了减少了cpu因线程等待的可能，让CPU一直处于工作状态。换句话说，如果我们能想办法减少CPU空闲时间，我们的计算机就可以支持更多的线程。  
其实线程是一个抽象概念，我们从物理层面理解，就是单位时间内把每毫核分配处理不同的任务，从而提高单位时间内CPU的利用率。

> 线程就是为了能自动分配CPU时间片而生的。

异步实现里面还是要用线程池限制一下线程数吧，否则没有达到减少线程的效果。

异步模式设计的程序可以显著减少线程等待，从而在高吞吐量的场景中，极大提升系统的整体性能，显著降低时延。

因此，像MQ这种需要超高吞吐量和超低时延的中间件系统，在其核心流程中，会大量采用异步设计。

# 1 案例引入 #

一个转账的微服务Transfer( accountFrom, accountTo, amount)，该服务有三参数

*  转出账户
 *  转入账户
 *  转账金额

要从账户A中转账100到账户B：

1.  先从A的账户中减去100元
2.  再给B的账户加上100元，转账完成。

*  对应时序图：  
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTg5NTEw_size_1_color_FFFFFF_t_70][]

调用另外一个微服务Add(account, amount)，给账户account增加金额amount，当amount为负值时，就是扣减相应金额。

> 为简化，省略了错误处理和事务相关代码

# 2 同步的性能瓶颈 #

同步实现，对应的伪代码：

Transfer(accountFrom, accountTo, amount) { 
      // 先从accountFrom的账户中减去相应的钱数
      Add(accountFrom, -1 * amount)
      // 再把减去的钱数加到accountTo的账户中
      Add(accountTo, amount)
      return OK
    }

先从accountFrom的账户中减去相应的钱数，再把减去的钱数加到accountTo的账户中，同步实现简单直接。  
性能如何？  
假设微服务Add的平均响应时延是50ms，微服务Transfer的平均响应时延大约等于执行2次Add的时延，即100ms。随调用Transfer服务的请求越来越多，会出现什么情况呢？

该实现，每处理一个请求耗时100ms，这100ms过程要独占个线程。  
可得：每个线程每秒最多可处理10个请求。假设服务器同时打开的线程数量上限10,000，可以计算出这台服务器每秒钟可以处理的请求上限是： 10,000 （个线程）\* 10（次请求每秒） = 100,000 次每秒。

若请求速度超过这个值，那么请求就不能被马上处理，只能阻塞或者排队，这时候Transfer服务的响应时延由100ms延长到了：排队的等待时延 + 处理时延(100ms)。即大量请求时，我们的微服务的平均响应时延变长了。

这是不是已达服务器极限？远没有！  
若监测服务器指标，会发现无论是CPU、内存，还是网卡流量或者是磁盘的IO都空闲的很，那我们Transfer服务中的那10,000个线程在作甚？  
没错！绝大部分线程都在等待Add服务返回结果。

即采用同步，整个服务器的所有线程大部分时间都没在工作，而在等待！

若能减少或避免这种无意义等待，即可大幅提升服务吞吐能力，提升性能。

# 异步实现方案 #

异步实现同样的业务。

TransferAsync(accountFrom, accountTo, amount, OnComplete()) { 
      // 异步从accountFrom的账户中减去相应的钱数，然后调用OnDebit方法。
      AddAsync(accountFrom, -1 * amount, OnDebit(accountTo, amount, OnAllDone(OnComplete())))
    }
    // 扣减账户accountFrom完成后调用
    OnDebit(accountTo, amount, OnAllDone(OnComplete())) { 
      // 再异步把减去的钱数加到accountTo的账户中，然后执行OnAllDone方法
      AddAsync(accountTo, amount, OnAllDone(OnComplete()))
    }
    // 转入账户accountTo完成后调用
    OnAllDone(OnComplete()) { 
      OnComplete()
    }

TransferAsync服务比Transfer多个参数，且该参数传入的是一个回调方法OnComplete()（虽然Java并不支持将方法作为方法参数传递，但像JavaScript等很多语言都具有这样特性，Java可传个回调类的实例来变相实现）。

*  TransferAsync()方法语义  
    请帮我执行转账操作，当转账完成后，请调用OnComplete()方法。  
    调用TransferAsync的线程不必等待转账完成即可立即返回，待转账结束，TransferService自然会调用OnComplete()方法来执行转账后续工作。

异步实现相对同步稍复杂。先定义俩回调方法：

1.  OnDebit()：扣减账户accountFrom完成后调用的回调方法
2.  OnAllDone()：转入账户accountTo完成后调用的回调方法。

异步实现语义相当于：

1.  异步从accountFrom的账户中减去相应钱数，然后调用OnDebit
2.  在OnDebit方法中，异步把减去的钱数加到accountTo的账户中，然后执行OnAllDone
3.  在OnAllDone中调用OnComplete

*  时序图  
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTg5NTEw_size_1_color_FFFFFF_t_70 1][]

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

## 异步实现的性能 ##

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

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

## 异步框架: CompletableFuture ##

开发时可使用异步框架和响应式框架，解决一些异步编程问题。  
Java中比较常用的异步框架有Java8内置的CompletableFuture和ReactiveX的RxJava。

*  CompletableFuture简单实用易理解
 *  RxJava功能更强大

Java 8中新增的异步编程类：CompletableFuture，几乎囊获了开发异步程序的大部分功能，使用CompletableFuture很容易编写优雅且易维护的异步代码。

接下来用CompletableFuture实现的转账服务。

CompletableFuture定义2个微服务的接口：

/** * 账户服务 */
    public interface AccountService { 
        /** * 变更账户金额 * @param account 账户ID * @param amount 增加的金额，负值为减少 */
        CompletableFuture<Void> add(int account, int amount);
    }

/** * 转账服务 */
    public interface TransferService { 
        /** * 异步转账服务 * @param fromAccount 转出账户 * @param toAccount 转入账户 * @param amount 转账金额，单位分 */
        CompletableFuture<Void> transfer(int fromAccount, int toAccount, int amount);
    }

接口中定义的方法的返回类型都是泛型CompletableFeture，泛型类型就是真正方法需要返回数据的类型，这俩服务无需返回数据，所以用Void。

实现转账服务：

/** * 转账服务的实现 */
    public class TransferServiceImpl implements TransferService { 
        @Inject
        private  AccountService accountService; // 使用依赖注入获取账户服务的实例
        @Override
        public CompletableFuture<Void> transfer(int fromAccount, int toAccount, int amount) { 
          // 异步调用add方法从fromAccount扣减相应金额
          return accountService.add(fromAccount, -1 * amount)
          // 然后调用add方法给toAccount增加相应金额
          .thenCompose(v -> accountService.add(toAccount, amount));    
        }
    }

先定义一个AccountService实例，这个实例从外部注入进来，至于怎么注入不是我们关心的问题，就假设这个实例是可用的就好了。

看transfer()方法，先调用一次账户服务accountService.add()方法从fromAccount扣减相应金额，因add()方法返回的就是个CompletableFeture对象，可用CompletableFeture的thenCompose()方法将下一次调用accountService.add()串联起来，实现异步依次调用两次账户服务完整转账。

客户端使用CompletableFuture也灵活，既可同步调用，也可异步。

public class Client { 
        @Inject
        private TransferService transferService; // 使用依赖注入获取转账服务的实例
        private final static int A = 1000;
        private final static int B = 1001;
    
        public void syncInvoke() throws ExecutionException, InterruptedException { 
            // 同步调用
            transferService.transfer(A, B, 100).get();
            System.out.println("转账完成！");
        }
    
        public void asyncInvoke() { 
            // 异步调用
            transferService.transfer(A, B, 100)
                    .thenRun(() -> System.out.println("转账完成！"));
        }
    }

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

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

# 总结 #

异步的思想就是，当要执行一项比较耗时的操作时，不去等待操作结束，而是给这个操作一个命令：“当操作完成后，接下来去执行什么。”

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

同时我们也需注意异步模型问题：相比同步，异步实现的复杂度大很多，代码可读性和可维护性都显著下降。  
虽然使用一些异步编程框架会在一定程度上简化异步开发，但是并不能解决异步模型高复杂度的问题。

异步性能虽好，勿滥用，只有类似MQ这种业务逻辑简单且需超高吞吐量场景，或须长时等待资源，才考虑使用异步模型。  
如果系统的业务逻辑比较复杂，在性能足够满足业务需求的情况下，采用符合人类自然的思路且易于开发和维护的同步模型是更加明智的选择。

# 面试场景题快问快答 #

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

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

在异步实现中，回调方法OnComplete()是在什么线程中运行的？我们是否能控制回调方法的执行线程数  
CompletableFuture默认是在ForkjoinPool commonpool里执行的，也可以指定一个Executor线程池执行，借鉴guava的ListenableFuture的时间，回调可以指定线程池执行，这样就能控制这个线程池的线程数目了。  
在异步实现中，回调方法 OnComplete()在执行OnAllDone()回调方法的那个线程，可通过一个异步线程池控制回调方法的线程数，如Spring中的async就是通过结合线程池来实现异步。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTg5NTEw_size_1_color_FFFFFF_t_70]: /images/20221119/6bc378eb57044a038acf32abf5e41067.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTg5NTEw_size_1_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200805222951345.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTg5NTEw,size_1,color_FFFFFF,t_70