熔断器 Hystrix 源码解析 —— 执行结果缓存

摘要: 原创出处 http://www.iocoder.cn/Hystrix/command-execute-result-cache/ 「芋道源码」欢迎转载，保留摘要，谢谢！

微信排版总是蹦蹦的，请点击【阅读原文】，抱歉~

本文主要基于 Hystrix 1.5.X 版本

• 1. 概述

• 2. 好处

• 3. Observable#defer(...)

• 4. AbstractCommand#toObservavle(...)

• 5. HystrixCachedObservable

• 6. HystrixCommandResponseFromCache

• 666. 彩蛋

---

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

4. 新的源码解析文章实时收到通知。每周更新一篇左右。

5. 认真的源码交流微信群。

---

1. 概述

本文主要分享 Hystrix 执行命令的结果缓存。

建议 ：对 RxJava 已经有一定的了解的基础上阅读本文。

Hystrix 执行命令整体流程如下图：

FROM 《【翻译】Hystrix文档-实现原理》「流程图」

• 红圈 ：在 《Hystrix 源码解析 —— 执行命令方式》 有详细解析。

• 紫圈 ：在 #toObservable() 方法里，如果请求结果缓存这个特性被启用，并且缓存命中，则缓存的回应会立即通过一个 Observable 对象的形式返回；如果缓存未命中，则返回【订阅了执行命令的 Observable】的 ReplySubject 对象缓存执行结果。

• ReplySubject 能够重放执行结果，从而实现缓存的功效。本文不对 ReplySubject 做太多拓展，感兴趣的同学可以阅读 《ReactiveX/RxJava文档中文版 —— Subject》 。

在官方提供的示例中，我们使用 CommandUsingRequestCache 进行调试 。

推荐 Spring Cloud 书籍：

• 请支持正版。下载盗版，等于主动编写低级 BUG 。

• 程序猿DD —— 《Spring Cloud微服务实战》

• 周立 —— 《Spring Cloud与Docker微服务架构实战》

• 两书齐买，京东包邮。

2. 好处

点击 《【翻译】Hystrix文档-实现原理》「请求缓存」 ，查看对请求缓存的好处分享，写的真的很赞。

3. Observable#defer(...)

本小节为拓展内容，源码解析 RxJava ( 非 Hystrix ) 的 Observable#defer(...) 的方法实现。考虑到 Hystrix 大量使用，为了更好的理解，解析下源码。

《RxJava 源码解析 —— Observable#defer(...)》

4. AbstractCommand#toObservavle(...)

AbstractCommand#toObservavle(...) 方法，代码如下 ：

1.  1: public Observable<R> toObservable() {

2.  2:     final AbstractCommand<R> _cmd = this;

3.  3:

4.  4:     //doOnCompleted handler already did all of the SUCCESS work

5.  5:     //doOnError handler already did all of the FAILURE/TIMEOUT/REJECTION/BAD_REQUEST work

6.  6:     final Action0 terminateCommandCleanup = new Action0() {} // ... 省略

7.  7:

8.  8:     //mark the command as CANCELLED and store the latency (in addition to standard cleanup)

9.  9:     final Action0 unsubscribeCommandCleanup = new Action0() {} // ... 省略

10. 10:

11. 11:     final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {

12. 12:         @Override

13. 13:         public Observable<R> call() {

14. 14:             if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {

15. 15:                 return Observable.never();

16. 16:             }

17. 17:             return applyHystrixSemantics(_cmd);

18. 18:         }

19. 19:     };

20. 20:

21. 21:     final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() {} // ... 省略

22. 22:

23. 23:     final Action0 fireOnCompletedHook = new Action0() {} // ... 省略

24. 24:

25. 25:     return Observable.defer(new Func0<Observable<R>>() {

26. 26:         @Override

27. 27:         public Observable<R> call() {

28. 28:             /* this is a stateful object so can only be used once */

29. 29:             if (!commandState.compareAndSet(CommandState.NOT_STARTED, CommandState.OBSERVABLE_CHAIN_CREATED)) {

30. 30:                 IllegalStateException ex = new IllegalStateException("This instance can only be executed once. Please instantiate a new instance.");

31. 31:                 //TODO make a new error type for this

32. 32:                 throw new HystrixRuntimeException(FailureType.BAD_REQUEST_EXCEPTION, _cmd.getClass(), getLogMessagePrefix() + " command executed multiple times - this is not permitted.", ex, null);

33. 33:             }

34. 34:

35. 35:             // 命令开始时间戳

36. 36:             commandStartTimestamp = System.currentTimeMillis();

37. 37:

38. 38:             // TODO【2001】【打印日志】

39. 39:             if (properties.requestLogEnabled().get()) {

40. 40:                 // log this command execution regardless of what happened

41. 41:                 if (currentRequestLog != null) {

42. 42:                     currentRequestLog.addExecutedCommand(_cmd);

43. 43:                 }

44. 44:             }

45. 45:

46. 46:             // 缓存开关、缓存KEY

47. 47:             final boolean requestCacheEnabled = isRequestCachingEnabled();

48. 48:             final String cacheKey = getCacheKey();

49. 49:

50. 50:             // 优先从缓存中获取

51. 51:             /* try from cache first */

52. 52:             if (requestCacheEnabled) {

53. 53:                 HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);

54. 54:                 if (fromCache != null) {

55. 55:                     isResponseFromCache = true; // 标记 从缓存中结果

56. 56:                     return handleRequestCacheHitAndEmitValues(fromCache, _cmd);

57. 57:                 }

58. 58:             }

59. 59:

60. 60:             // 获得 执行命令Observable

61. 61:             Observable<R> hystrixObservable =

62. 62:                     Observable.defer(applyHystrixSemantics)

63. 63:                             .map(wrapWithAllOnNextHooks);

64. 64:

65. 65:             // 获得 缓存Observable

66. 66:             Observable<R> afterCache;

67. 67:             // put in cache

68. 68:             if (requestCacheEnabled && cacheKey != null) {

69. 69:                 // wrap it for caching

70. 70:                 HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);

71. 71:                 // 并发若不存在

72. 72:                 HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.putIfAbsent(cacheKey, toCache);

73. 73:                 if (fromCache != null) { // 添加失败

74. 74:                     // another thread beat us so we'll use the cached value instead

75. 75:                     toCache.unsubscribe();

76. 76:                     isResponseFromCache = true; // 标记 从缓存中结果

77. 77:                     return handleRequestCacheHitAndEmitValues(fromCache, _cmd);

78. 78:                 } else { // 添加成功

79. 79:                     // we just created an ObservableCommand so we cast and return it

80. 80:                     afterCache = toCache.toObservable();

81. 81:                 }

82. 82:             } else {

83. 83:                 afterCache = hystrixObservable;

84. 84:             }

85. 85:

86. 86:             //

87. 87:             return afterCache

88. 88:                     .doOnTerminate(terminateCommandCleanup)     // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line))

89. 89:                     .doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once

90. 90:                     .doOnCompleted(fireOnCompletedHook);

91. 91:         }

92. 92:     });

93. 93: }

• 第 2 行 ： _cmd 指向当前命令对象，用于下面实现 FuncX ，ActionX 内部类使用。

• 第 11 至 19 行 ：当缓存特性未开启，或者缓存未命中时，使用 applyHystrixSemantics 传入 Observable#defer(...) 方法，声明执行命令的 Observable。

• 第 25 行 ：声明缓存 Observable 。Hystrix 执行命令的 Observable 声明关系如下：

• 第 29 至 33 行 ：一条命令只能执行一次。

• 第 36 行 ：记录命令开始时间戳。

• 第 38 至 44 行 ：TODO【2001】【打印日志】

• 第 47 至 48 行 ：缓存存开关、KEY 。

• 第 52 至 58 行 ：如果请求结果缓存这个特性被启用，并且缓存命中，则缓存的回应会立即通过一个 Observable 对象的形式返回。

• 第 53 行 ： requestCache 缓存，在 TODO 【2008】【请求缓存】 详细解析。

• 第 53 行 ：「6. HystrixCommandResponseFromCache」 详细解析。

• 第 56 行 ： #handleRequestCacheHitAndEmitValues(...) 方法，在第 78 行详细解析。

• 第 61 至 63 行 ：获取执行命令的 Observable 。在 《Hystrix 源码解析 —— 命令执行（一）之正常执行逻辑》 详细解析。

• 第 68 至 81 行 ：当缓存特性开启，并且缓存未命中时，创建【订阅了执行命令的 Observable】的 HystrixCommandResponseFromCache 。

• 第 75 行 ：调用 HystrixCommandResponseFromCache#unsubscribe() 方法，取消 HystrixCommandResponseFromCache 的订阅。这一步很关键，因为我们不希望缓存不存在时，多个线程去执行命令，最好有且只有一个线程执行命令。在 「5. HystrixCachedObservable」 详细解析。

• 第 77 行 ：「6. HystrixCommandResponseFromCache」 详细解析。

• 第 69 至 72 行 ：创建 HystrixCommandResponseFromCache ，并添加到 requestCache 。哟， HystrixRequestCache#putIfAbsent(...) 方法，多个线程添加时，只有一个线程添加成功。

• 第 73 至 77 行 ：添加失败的线程( 们 )：

• 第 80 行 ：调用 HystrixCommandResponseFromCachetoObservable() 方法，获得缓存 Observable 。

• 第 82 至 84 行 ：当缓存特性未开启，使用执行命令 Observable 。

• 第 87 至 91 行 ：在返回的 Observable 上，订阅一些清理的处理逻辑。对这几个方法有疑惑的同学，可以阅读 《给 Android 开发者的 RxJava 详解》「 3) Subscribe (订阅) 」 。

5. HystrixCachedObservable

com.netflix.hystrix.HystrixCachedObservable ，缓存 Observable 。

HystrixCachedObservable 构造方法，代码如下 ：

1.  1: public class HystrixCachedObservable<R> {

2.  2: /**

3.  3:  * 订阅

4.  4:  */

5.  5: protected final Subscription originalSubscription;

6.  6: /**

7.  7:  * 缓存 cachedObservable

8.  8:  */

9.  9: protected final Observable<R> cachedObservable;

10. 10: /**

11. 11:  * TODO 【2006】【outstandingSubscriptions】

12. 12:  */

13. 13: private volatile int outstandingSubscriptions = 0;

14. 14: //private AtomicInteger outstandingSubscriptions2 = new AtomicInteger(0);

15. 15:

16. 16: protected HystrixCachedObservable(final Observable<R> originalObservable) {

17. 17:     ReplaySubject<R> replaySubject = ReplaySubject.create();

18. 18:     this.originalSubscription = originalObservable

19. 19:             .subscribe(replaySubject);

20. 20:

21. 21:     this.cachedObservable = replaySubject

22. 22:             .doOnUnsubscribe(new Action0() {

23. 23:                 @Override

24. 24:                 public void call() {

25. 25:                     outstandingSubscriptions--;

26. 26:                     if (outstandingSubscriptions == 0) {

27. 27:                         originalSubscription.unsubscribe();

28. 28:                     }

29. 29:                 }

30. 30:             })

31. 31:             .doOnSubscribe(new Action0() {

32. 32:                 @Override

33. 33:                 public void call() {

34. 34:                     outstandingSubscriptions++;

35. 35:                 }

36. 36:             });

37. 37: }

• 第 17 至 19 行 ：实际上，HystrixCachedObservable 不是一个 Observable 的子类，而是对传入的 Observable 封装 ：使用 ReplaySubject 向传入的 Observable 发起订阅，通过 ReplaySubject 能够重放执行结果，从而实现缓存的功效。这里有几个卡到笔者的并且很有趣的点，我们一一道来 ：从上文中，我们可以看到，传入的 originalObservable 为 hystrixObservable 执行命令 Observable 。在 Hystrix 里，提供了两种执行命令的隔离方式 ：线程池( THREAD ) 和信号量( SEMAPHORE )。

• 当使用 THREAD 隔离时， #subscribe(replaySubject) 调用完成时，实际命令并未开始执行，或者说，这是一个异步的执行命令的过程。那么，会不会影响返回执行结果呢？答案当然是不会，BlockingObservable 在得到执行完成才会结束阻塞，此时已经有执行结果。

• 当使用 SEMAPHORE 隔离时， #subscribe(replaySubject) 调用完成时，实际命令已经执行完成，所以即使 AbstractCommand#toObservavle(...) 的第 75 行 ：调用 HystrixCommandResponseFromCache#unsubscribe() 方法，也会浪费，重复执行命令。而对于 THREAD 隔离的情况，通过取消订阅的方式，只会执行一次命令。当然，如果“恶搞” THREAD 隔离的情况，增加 sleep 的调用如下，就能达到重复执行命令的效果。

• 第 21 至 36 行 ：TODO 【2006】【outstandingSubscriptions】原子性没问题么？历史版本使用的是 AtomicInteger 。

---

HystrixCachedObservable 的其他方法，点击 链接 查看。

6. HystrixCommandResponseFromCache

com.netflix.hystrix.HystrixCommandResponseFromCache ，是 HystrixCachedObservable 的子类。在父类的基础上，增加了对 AbstractCommand.executionResult 的关注。

HystrixCachedObservable#from(Observable, AbstractCommand) 方法，创建 HystrixCommandResponseFromCache 对象，点击 链接 查看。

---

HystrixCommandResponseFromCache#toObservableWithStateCopiedInto(...) 方法，点击 链接 查看。

• 通过 completionLogicRun 属性，保证 #doOnError() ， #doOnCompleted() ， #doOnUnsubscribe() 方法有且只有一个方法执行具体逻辑。

• #doOnError() ， #doOnCompleted() 执行时，调用 #commandCompleted() 方法，从缓存命令( HystrixCommandResponseFromCache.originalCommand ) 复制 executionResult 属性给当前命令( commandToCopyStateInto ) 。

• #doOnUnsubscribe() 执行时，调用 #commandUnsubscribed() 方法，使用当前命令( commandToCopyStateInto )自己的 executionResult ，不进行复制。

• TODO 【2007】【executionResult】用途

666. 彩蛋

如鲠在喉的感觉，从周六开始磨了四天多，一直没写到一个比较舒服的状态。

先发现发，如果有不清晰的地方，烦请指出，谢谢。

胖友，分享一波朋友圈可好！