京东架构师手把手教你如何使用异步编程，提升服务性能
Posted 京东技术
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了京东架构师手把手教你如何使用异步编程，提升服务性能相关的知识，希望对你有一定的参考价值。
来这里找志同道合的小伙伴！
庞大复杂的系统通常会采用服务化组件来实现。系统越复杂，组件之间的依赖和调用关系也会越复杂。对于处于底层的基础服务，直接和间接的调用所带来的流量压力非常大。处于中间层的聚合型服务，面对的挑战则是依赖的服务太多，后端个别服务的性能延迟就会影响其吞吐量。性能优化是我们系统稳定性中的重要一环，这其中，调用所依赖的RPC服务或后端数据是重点之一。目前，除了传统JDBC这样从API到主流驱动实现就是阻塞式的类库之外，其他常用的RPC/HTTP服务、MQ、Redis、Mongodb、Kafka等系统都提供了成熟的基于NIO的客户端库，也有相应的异步API。但是目前交易平台的大多数中台服务系统，还在习惯性使用着这些库的同步API，并不能充分的利用CPU，这也给我们带来了一定的优化空间。从16年开始我们在一些核心的但是服务逻辑相对简单的系统中使用异步方式来实现，虽然暂时还做不到完全的异步化，但是也取得了比较好的效果。这篇文章虽然更多是一个简介性质，但是也涵盖了我们在异步编程中需要关注的要点。希望大家能够习惯和拥抱异步编程。
一、相关概念介绍
同步(Synchronous)/异步(Asynchronous)，通常是指函数调用中的消息通信的两种不同模式。
1、异步和同步的区别
函数调用发生时，消息(参数)从caller传递到callee，控制权(指令执行)从caller转移到callee。调用返回时，控制权从callee转移到caller。两者的区别在于，callee是否需要等待执行完成才将控制权转移给caller。
在RPC这种更复杂的场景下，本质上并没有不同。
◆ 同步
1.callee执行完成才返回
2.返回值即结果
◆ 异步
1.callee不需要执行完成就可返回
2.caller要获取结果，需要通过轮询、回调等机制
◆ 同步RPC
◆ 异步RPC
可以看到，在异步RPC的场景下，客户端和服务端用于处理IO的CPU能得到充分利用，通常只需要远低于caller请求数量的线程就可以了，这就是多路复用。
2、callee执行机制
上图中callee的background execute通常是采用池化线程来完成的，比如ThreadPoolExecutor或EventLoop1。
3、caller获取执行结果
caller调用callee时，如果需要获取执行结果（消息双向传递），或者获知执行是否完成（消息单向传递无返回值），在异步模式下，主要依靠下面两种机制。
◆ 轮询(Polling)
比如Java的Future就提供了isDone()这种询问机制。
1.//Caller.java
2.void call() {
3. Future<Void> f = callee.asyncCall(param);
4. // do some other things
5. while(true) {
6. if (f.isDone()) break;
7. //do some other things or sleep or timeout
8. }
9.}
或阻塞版本
1.//Caller.java
2.void call() {
3. Future<Void> f = callee.asyncCall(param);
4. // do some other things
5. f.get(timeout, TimeUnit.SECONDS);
6.}
轮询的控制逻辑在caller端。
◆ 回调(Callback)
caller设置一个回调函数，供callee执行完成后调用这个函数。回调的控制是反转的，通常由callee端控制。
1.//Caller.java
2.void call() {
3. callee.asyncCall(param, new AsyncHandler<Response<Message>>() {
4. @Override public void handleResponse(Response<Message> response) {
5. msg = response.get();
6. // process the msg...
7. }
8. });
9. // do some other things
10.}
4、异步模式的场景
◆ 阻塞
阻塞(Blocking)/非阻塞(Non-Blocking)是用来描述，在等待调用结果时caller线程的状态。阻塞，通常意味着caller线程不再使用CPU时间，处于可被OS调度的状态(注意与Java线程状态2的区别)。 磁盘IO和网络IO是常见的会引起线程阻塞的场景3。受制于底层OS的同步阻塞式IO系统函数，调用Java OIO(Old blocking IO) API无疑是会阻塞的。对于DiskIO，Java NIO2提供了异步API。对于SocketIO，Java NIO2以及NIO框架Netty，都提供了异步API。
◆ Linux提供了异步IO系统函数，只能用于DiskIO，还有一些限制4，Java NIO2 AsynchronousFileChannel内部仍然使用线程池+阻塞式API的实现。
◆ Linux为SocketIO准备就绪阶段提供了非阻塞式API(select/poll/epoll)，但是IO执行阶段仍然是同步阻塞的，因此主流的Java NIO框架的Reactor模式内部实现使用了线程池。
◆ 并行
比如需要调用多个没有依赖关系的服务，或者访问分散在多个存储分片中的数据，如果服务接口或数据访问接口实现了异步API，那么就很方便实现并行调用，减少总体调用耗时。
◆ 速度不匹配
使用中间队列解偶caller和callee的速度不匹配问题，削峰填谷。
◆ 批量
使用中间队列解偶caller和callee的速度不匹配问题，削峰填谷。
二、异步API的几种风格
1、Callback
这个比较传统，比如zookeeper客户端提供的基于回调的异步API:
1.try {
2. zookeeper.create(path, data, acl, createMode, new StringCallback() {
3. public void processResult(int rc, String path, Object ctx, String name) {
4. if (rc != KeeperException.Code.OK.intValue()) {
5. // error handle
6. } else {
7. // success process
8. // 如果需要在成功后再发起基于回调的异步调用，会形成callback hell
9. }
10. }
11. }, ctx);
12.} catch( Throwable e) {
13. // error handle
◆ Callback通常是无状态的
◆ 要获取Callback的计算结果，通常需要closure
◆ 异常处理比较分散
◆ 在有多个异步调用链的时候，容易造成Callback hell
2、Future/Promise
Promise是callee给caller的凭证，代表未完成但承诺完成（成功或失败）的结果。Promise本身是有状态的，通常由callee端维护。其状态转移如下（术语参考Promise/A+和ES6）：
Promise的状态只能转移一次，因此如果有callback，那么.then(callback)或.catch(callback)也只被执行一次。
JDK5的Future只能用轮询或者阻塞的方式获取结果，caller端处理比较繁琐。Guava的ListenableFuture，特别是JDK8的CompletableFuture，则是完整实现了Promise风格的异步API。 个人认为Promise是更好的Callback，ListenableFuture接口只是比Future多了一个void addListener(Runnable, Executor)方法。 Promise提供了比Callback更易用更清晰的编程模式，尤其是涉及多个异步API的串行调用（chaining或pipelining )、组合调用（并行、合并）、异常处理等方面有很大的优势。
引申阅读
◆ 这篇文章谈到了Future和Promise的细微区别、相关历史和技术。
◆ 这里有一些讨论：Aren’t promises just callbacks?,Is there really a fundamental difference between callbacks and Promises?。
◆ Promise借鉴了函数式中的一些概念: 从函数式编程到Promise。
◆ 这篇文章简要对比了几种语言中的Promise框架。
3、ReactiveX
其官方网站的介绍
An API for asynchronous programming with observable streams
其关键的概念Observable比较Promise来说：
◆ Promise代表一个异步计算值，而Observable代表着一系列值(stream)。
◆ Promise的值只能产生一次，而Observable的事件可以不断产生。因此Rx首先流行在前端UI场景：事件来源多，数据变化影响多个UI组件的变更。
Rx的学习曲线比Promise要高得多，而且目前Promise风格的异步编程能够满足我们大部分的服务端开发场景，因此我们这里主要关注Promise。
三、Promise在服务端的应用
下面穿插着以JDK8的CompletableFuture和Guava的ListenableFuture（适用JDK6）为例介绍Promise的用法。
1、符合Promise风格的方法签名
Promise风格的方法签名，有个不成文的规则是不抛出异常，因为异常是Promise对象本身就能携带的两种状态之一。比如我们想把一个Callback风格的异步API包装成Promise风格的（通常在使用一个较老的类库时需要这样的包装），可以这样：
1.//caller.java
2.CompletableFuture<String> asyncCall(final String msg) {
3. CompletableFuture<String> promise = new CompletableFuture<>();
4. try {
5. callee.asyncCall(msg, new Callback<String>() {
6. public void onSuccess(String r) { promise.complete(r); }
7. public void onFail(Throwable t) { promise.completeExceptionally(t); }
8. });
9. } catch (Throwable e) {
10. promise.completeExceptionally(t);
11. }
12. return promise;
13.}
下面是使用ListenableFuture的实现异步发送消息的API。
1.//LocalMessageEngine.java
2.static class WriteTask {
3. // SettableFuture是Guava中一种可设置状态的Promise类型。
4. final SettableFuture<Boolean> promise = SettableFuture.create();
5. final byte[] message;
6. WriteTask(byte[] message) {
7. this.message = message;
8. }
9.}
10.public Producer createProducer() {
11. return new Producer() {
12. public ListenableFuture<Boolean> asyncProduce(byte[] message) {
13. if (!Engine.this.started) {
14. // 返回前已完成
15. return Futures.immediateFailedFuture(new IllegalStateException("Message engine was stopped or not started"));
16. }
17.
18. WriteTask task = new WriteTask(message);
19. boolean queued = writeTaskQueue.offer(task);
20. if (!queued) {
21. task.promise.set(Boolean.FALSE);
22. }
23. return task.promise;
24. }
25. };
26.}
上面两个例子，描述了如何创建一个Promise对象返回给caller，以及如何在callee端fulfill或reject这个Promise。你可能会发现，返回给caller之前Promise是可以处于完成状态的。在继续下面的使用介绍前，先简单的看下ListenableFuture和CompletableFuture的几个主要API。
2、串行调用
ListenableFuture没有提供then方法，而是通过Futures的一系列静态方法来实现Promise风格的API。由于两者有大部分的API是可以相互转化的，限于篇幅下面就不全部演示了。Promse的callback就是其then或catch方法的函数型参数，当Promise被resolve时执行这个callback函数。这个callback函数的输入，就是Promise的resolved值。根据callback函数的输出的不同，需要采取不同的then方法。
◆ callback无输出
Futures.addCallback和CompletableFuture.thenAccept接受无输出的callback。
1.// Guava
2.ListenableFuture<QueryResult> promise1 = ...;
3.Futures.addCallback(promise1, new FutureCallback<QueryResult>() {
4. public void onSuccess(QueryResult result) {
5. storeInLocalCache(result);
6. }
7. public void onFailure(Throwable t) {
8. reportError(t);
9. }
10.});
1.// CompletableFuture
2.CompletableFuture<QueryResult> promise1 = ...;
3.CompletableFuture<Void> promise2 = promise1.thenAccept(result -> storeInLocalCache(result));
4.return promise2;
5.//return promise1.thenAccept(result -> storeInLocalCache(result));//thenAccept返回另一个Promise实例
先忽略异常处理。对比下这种场景下的ListenableFuture和CompletableFuture：前者采取了更传统的callback风格，后者则返回一个新的Promise实例，callback计算完毕则promise2被fulfilled，很容易通过promise2来获取callback执行完毕与否，不需要closure。
◆ callback输出一个普通计算值
这种情况下callback就是一个转换函数，输入是前一个Promise的fulfilled值，输出则作为新Promise的fulfilled值。Futures.transform和CompletableFuture.thenApply接收这样的callback函数。
1.// CompletableFuture
2.CompletableFuture<QueryResult> queryFuture = ...;
3.CompletableFuture<List<Row>> rowsFuture = queryFuture.thenApply(result -> result.getRows());
4.return rowsFuture;
◆ callback输出一个异步计算值，即一个Promise
乍一看，这种情况下的输出跟上一种好像没什么区别。但实际上，输出一个Promise值和输出一个普通的值有根本的区别。还记得吧，Promise代表着一个未完成的并且承诺完成的值。通常这种情况下，意味着callback里调用了另外一个Promise风格的异步API。比如下面的例子中indexService.lookUp和dataService.read方法，由于涉及到IO，都设计为异步API。
1.//Guava
2.ListenableFuture<RowKey> rowKeyFuture = indexService.lookUp(query);
3.AsyncFunction<RowKey, QueryResult> queryFunction = new AsyncFunction<RowKey, QueryResult>() {
4. public ListenableFuture<QueryResult> apply(RowKey rowKey) {
5. return dataService.read(rowKey);
6. }
7.};
8.ListenableFuture<QueryResult> queryFuture = Futures.transformAsync(rowKeyFuture, queryFunction);
9.return queryFuture;
Futures.transformAsync和CompletableFuture.thenCompose接收这样的callback函数。
设想一下，如果某个逻辑中需要调用的多个Promise风格的异步方法（比如多个RPC调用），并且有先后依赖关系，即上一个方法的执行结果作为下一个方法的输入。就可以用thenCompose把他们串起来。
1.//CompletableFuture
2.CompletableFuture<RPC4Result> promise4 = rpc1.call(input) //promise1
3. .thenCompose(rpc1Result -> rpc2.call(rpc1Result)) //promise2
4. .thenCompose(rpc2Result -> rpc3.call(rpc2Result)) //promise3
5. .thenCompose(rpc3Result -> rpc4.call(rpc3Result)) //promise4
6.return promise4;
不要被链式调用给忽悠了，你还是可以正常使用普通的风格。
单纯看来，上述的串行调用场景下使用Promise风格的API好像只是消除了Callback hell。那么采用同步API就既没有Callback hell的问题，又符合数据依赖关系。可是，你会发现，上面的举例中结尾都返回了Promise，就是说，包含这段代码的方法被设计为异步API。而使用同步API，则会强制这个方法的调用者只能使用同步方式调用。
3、并行调用
异步API很适合并行调用。caller在调用多个没有依赖关系的异步API时，可以先依次发起调用而不用等待每个调用真正执行完成，从callee的角度来讲，执行是并行的。caller可以对调用结果进行合并处理，关键是，合并也是异步风格的。
1.//Guava
2.List<ListenableFuture<QueryResult>> partialPromises = new ArrayList<ListenableFuture<QueryResult>>(nodes.size());
3.for (Node node : nodes) {
4. partialPromises.add(lookupHandler(node).query());
5.}
6.ListenableFuture<List<Row>> mergedPromise = Futures.transform(Futures.allAsList(partialPromises), new Function<List<List<Row>>, List<Row>>() {
7. @Override public Long apply(List<List<Row>> input) {
8. return merge(input);
9. }
10.})
11.return mergedPromise;
Futures.allAsList是并行执行所有的promises，若有一个promise异常完成则尝试reject尚未resolved的promise。也可以使用Futures.successfulAsList，区别在于后者并不会reject尚未resolved的promise。CompletableFuture的对应物是allOf和anyOf。
4、调用编排
合并结果设计为异步风格的好处在于，很方便做合并、串行混合调用编排，比如某个逻辑中需要调用四个个RPC服务A、B、C、D，其中：A的输出作为B、C的输入，B、C可并行，B、C的输出合并后作为D的输入。
1.//CompletableFuture
2.CompletableFuture<AResult> promiseA = rpcA.call(input);
3.CompletableFuture<DResult> promiseD = promiseA.thenCompose(aResult -> {
4. CompletableFuture<BResult> promiseB = rpcB.call(aResult);
5. CompletableFuture<CResult> promiseC = rpcC.call(aResult);
6. CompletableFuture<MergedResult> mergedPromise = promiseB.thenCombine(promiseC, (bResult, cResult) -> {
7. return merge(bResult, cResult)
8. });
9. return mergedPromise;
10.}).thenCompose(mergedResult -> rpcD.call(mergedResult));
11.return promiseD;
5、异常处理
上面提到过Callback风格的异步API，异常处理比较分散。而Promise风格的异常处理则优雅得多。我们需要记住，异常是Promise携带的两种状态之一。那么异常可以作为callback函数的输入。
◆ 通用异常处理
Futures.catching和CompletableFuture.exceptionally接收异常值为参数的callback函数。
1.//Guava
2.ListenableFuture<Integer> fetchCounterPromise = ...;
3.// Falling back to a zero counter in case an exception happens when
4.// processing the RPC to fetch counters.
5.ListenableFuture<Integer> faultTolerantPromise = Futures.catching(
6. fetchCounterPromise, FetchException.class,
7. new Function<FetchException, Integer>() {
8. public Integer apply(FetchException e) {
9. return 0;
10. }
11. });
我们再看一个更复杂的例子。
1.//CompletableFuture
2.CompletableFuture<FaultTolerantResult> faultTolerantPromise = rpc1.call(input) //promise1
3. .thenCompose(rpc1Result -> rpc2.call(rpc1Result)) //promise2
4. .thenCompose(rpc2Result -> rpc3.call(rpc2Result)) //promise3
5. .thenCompose(rpc3Result -> rpc4.call(rpc3Result)) //promise4
6. .exceptionally(err -> {
7. // process err
8. return faultTolerantValue;
9. }); //faultTolerantPromise
10.return faultTolerantPromise;
再提醒一遍，不要被链式调用迷惑了。这个例子里面，rpc1/rpc2/rpc3/rpc4都有可能发生异常，但我们只需要在最后统一处理（返回异常值或转换为一个默认正常值）。比如rpc2发生异常，那么rpc3/rpc4的逻辑（接收正常值的callback函数）都不会执行，但是rpc2的异常会传递给promise3/promise4。
◆ 恢复
假设在读取某个数据存储发生异常，我们需要某种恢复机制，比如读取另一个backup的数据存储（某种重试），那么可以使用Futures.cachingAsync和CompletableFuture.handle。
1.//CompletableFuture
2.public <V> CompletableFuture<V> dispatch(final Command<V> command) {
3. final CompletableFuture<V> dispatched = loadbalance.selectHandler().dispatch(command);
4. if (maxTries > 0) {
5. final AtomicInteger leftTries = new AtomicInteger(maxTries);
6. final BiFunction<CompletableFuture<V>, Throwable, CompletableFuture<V>> fallback = new BiFunction<CompletableFuture<V>, Throwable, CompletableFuture<V>>() {
7. @Override
8. public CompletableFuture<V> apply(CompletableFuture<V> input, Throwable cause) {
9. if (cause == null) return input;
10. if (cause instanceof RecoverableException && leftTries.getAndDecrement() > 0) {
11. final CompletableFuture<V> next = loadbalance.selectHandler().dispatch(new Command<V>(command.type, command.args));
12. return next.handle((v, err) -> err).thenCompose(err -> apply(next, err));
13. }
14. CompletableFuture<V> errFuture = new CompletableFuture<>();
15. errFuture.completeExceptionally(cause);
16. return errFuture;
17. }
18. };
19. return dispatched.handle((v, err) -> err).thenCompose(err -> fallback.apply(dispatched, err));
20. }
21. return dispatched;
22.}
◆ 超时
Promise是一个承诺完成（成功或失败）的结果，但是并不承诺完成时间。所以，通常需要一种超时机制，幸运的是ListenableFuture和CompletableFuture都实现了Future接口。
1.CompletableFuture<FaultTolerantResult> faultTolerantPromise = ...;
2.try {
3. FaultTolerantResult result = faultTolerantPromise.get(1000, TimeUnit.SECONDS);
4. // process result
5.} catch (TimeoutException e) {
6. //尝试取消执行尚未开始的callback函数
7. faultTolerantPromise.cancel(false);
8.}
四、Promise异步编程的注意点
异步编程比同步编程困难。异步编程通常主要解决一小部分问题，比如阻塞。Promise借鉴了函数式编程的风格，大量的逻辑会分散在各种callback函数来实现。因此对于习惯了同步编程的OO式或命令式编程风格的开发人员，需要一定的习惯时间。
上面谈到callee执行机制的时候，谈到了线程池，那么callee计算完成时，callback函数的执行通常是池中resolve Promise的线程执行。但是，如果caller在设置callback的时候，Promise已经完成，那么callback的执行线程则是caller线程。因此，请特别关注callback函数的执行线程的差别。请遵循：
◆ callback尽量轻量
◆ callback避免阻塞
◆ 否则请指定执行线程
如果callback执行了大量的计算，甚至执行了阻塞式操作，那么就很有可能阻塞住Promise的resolve线程，通常这类线程都是极少的，比如执行IO的EventLoop线程，有可能造成其他Promise得不到执行。
摘自ListenableFuture的文档：
Note: For fast, lightweight listeners that would be safe to execute in any thread, consider MoreExecutors.directExecutor. Otherwise, avoid it. Heavyweight directExecutor listeners can cause problems, and these problems can be difficult to reproduce because they depend on timing. For example:
◆ The listener may be executed by the caller of addListener. That caller may be a UI thread or other latency-sensitive thread. This can harm UI responsiveness.
◆ The listener may be executed by the thread that completes this Future. That thread may be an internal system thread such as an RPC network thread. Blocking that thread may stall progress of the whole system. It may even cause a deadlock.
◆ The listener may delay other listeners, even listeners that are not themselves directExecutor listeners.
我们也的确碰到过使用MoreExecutors.directExecutor时，由于编写了太过复杂的callback链，导致线程死锁的问题。
CompletableFuture和ListenableFuture都有指定callback执行线程的方法：
1.//使用内置的通用ForkJoin线程池
2.completableFuture.thenAcceptAsync(callback);
3.//使用指定的线程执行器
4.completableFuture.thenAcceptAsync(callback, executor);
5.//使用指定的线程执行器
6.Futures.transform(input, callback, executor);
正因为异步编程的复杂性，因此目前我们也尽量在业务逻辑相对简单的应用上进行异步化改造。后续，我们也会评估Quasar等协程框架。
---------------------END---------------------
下面的内容同样精彩
点击图片即可阅读
以上是关于京东架构师手把手教你如何使用异步编程，提升服务性能的主要内容，如果未能解决你的问题，请参考以下文章
Python架构师手把手教你Scrapy 对接 Docker