实用技巧：Hystrix传播ThreadLocal对象（两种方案）

Posted IT牧场

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了实用技巧：Hystrix传播ThreadLocal对象（两种方案）相关的知识，希望对你有一定的参考价值。

目前，Spring Cloud已在南京公司推广开来，不仅如此，深圳那边近期也要基于Spring Cloud新开微服务了。

于是，领导要求我出一套基于Spring Cloud的快速开发脚手架（近期开源）。在编写脚手架的过程中，也顺带总结一下以前在项目中遇到的问题：

使用Hystrix时，如何传播ThreadLocal对象？

我们知道，Hystrix有隔离策略：THREAD以及SEMAPHORE。

如果你不知道Hystrix的隔离策略，可以阅读我的书籍《Spring Cloud与Docker微服务架构实战》，或者参考文档：https://github.com/Netflix/Hystrix/wiki/Configuration#executionisolationstrategy

引子

当隔离策略为 THREAD 时，是没办法拿到 ThreadLocal 中的值的。

举个例子，使用Feign调用某个远程API，这个远程API需要传递一个Header，这个Header是动态的，跟你的HttpRequest相关，我们选择编写一个拦截器来实现Header的传递（当然也可以在Feign Client接口的方法上加 RequestHeader ）。

示例代码：

 
   
   
 
    
  
    
    
  
  1. public class KeycloakRequestInterceptor implements RequestInterceptor {
    2. 
  
    
    
  
    3. 
  
    
    
  
  3.     private static final String AUTHORIZATION_HEADER = "Authorization";
    4. 
  
    
    
  
    5. 
  
    
    
  
  5.     @Override
    6. 
  
    
    
  
  6.     public void apply(RequestTemplate template) {
    7. 
  
    
    
  
  7.         ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
    8. 
  
    
    
  
  8.         Principal principal = attributes.getRequest().getUserPrincipal();
    9. 
  
    
    
  
    10. 
  
    
    
  
  10.         if (principal != null && principal instanceof KeycloakPrincipal) {
    11. 
  
    
    
  
  11.             KeycloakSecurityContext keycloakSecurityContext = ((KeycloakPrincipal) principal)
    12. 
  
    
    
  
  12.                     .getKeycloakSecurityContext();
    13. 
  
    
    
  
    14. 
  
    
    
  
  14.             if (keycloakSecurityContext instanceof RefreshableKeycloakSecurityContext) {
    15. 
  
    
    
  
  15.                 RefreshableKeycloakSecurityContext.class.cast(keycloakSecurityContext)
    16. 
  
    
    
  
  16.                         .refreshExpiredToken(true);
    17. 
  
    
    
  
  17.                 template.header(AUTHORIZATION_HEADER, "Bearer " + keycloakSecurityContext.getTokenString());
    18. 
  
    
    
  
  18.             }
    19. 
  
    
    
  
    20. 
  
    
    
  
  20.         }
    21. 
  
    
    
  
  21.         // 否则啥都不干
    22. 
  
    
    
  
  22.     }
    23. 
  
    
    
  
  23. }
    24.

你可能不知道Keycloak是什么，不过没有关系，相信这段代码并不难阅读，该拦截器做了几件事：

  • 使用 RequestContextHolder.getRequestAttributes() 静态方法获得Request。

  • 从Request获得当前用户的身份，然后使用Keycloak的API拿到Token，并扔到Header里。

  • 这样，Feign使用这个拦截器时，就会用你这个Header去请求了。

注：Keycloak是一个非常容易上手，并且功能强大的单点认证平台。

现实很骨感

以上代码可完美运行——但仅限于Feign不开启Hystrix支持时。

注：Spring Cloud Dalston以及更高版可使用 feign.hystrix.enabled=true 为Feign开启Hystrix支持。

当Feign开启Hystrix支持时，

 
   
   
 
    
  
    
    
  
  1. ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
    2.

是 null 。

原因在于，Hystrix的默认隔离策略是 THREAD 。而 RequestContextHolder 源码中，使用了两个血淋淋的 ThreadLocal 。

解决方案一：调整隔离策略

将隔离策略设为SEMAPHORE即可：

 
   
   
 
    
  
    
    
  
  1. hystrix.command.default.execution.isolation.strategy: SEMAPHORE
    2.

这样配置后，Feign可以正常工作。

但该方案不是特别好。原因是Hystrix官方强烈建议使用THREAD作为隔离策略！ 参考文档：

Thread or Semaphore

The default, and the recommended setting, is to run HystrixCommands using thread isolation ( THREAD) and HystrixObservableCommands using semaphore isolation ( SEMAPHORE).

Commands executed in threads have an extra layer of protection against latencies beyond what network timeouts can offer.

Generally the only time you should use semaphore isolation for HystrixCommands is when the call is so high volume (hundreds per second, per instance) that the overhead of separate threads is too high; this typically only applies to non-network calls.

于是，那么有没有更好的方案呢？

解决方案二：自定义并发策略

既然Hystrix不太建议使用SEMAPHORE作为隔离策略，那么是否有其他方案呢？答案是自定义并发策略，目前，Spring Cloud Sleuth以及Spring Security都通过该方式传递 ThreadLocal 对象。

下面我们来编写自定义的并发策略。

编写自定义并发策略

编写自定义并发策略比较简单，只需编写一个类，让其继承 HystrixConcurrencyStrategy ，并重写 wrapCallable 方法即可。

代码示例：

 
   
   
 
    
  
    
    
  
  1. @Component
    2. 
  
    
    
  
  2. public class RequestAttributeHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
    3. 
  
    
    
  
  3.     private static final Log log = LogFactory.getLog(RequestHystrixConcurrencyStrategy.class);
    4. 
  
    
    
  
    5. 
  
    
    
  
  5.     public RequestHystrixConcurrencyStrategy() {
    6. 
  
    
    
  
  6.         HystrixPlugins.reset();
    7. 
  
    
    
  
  7.         HystrixPlugins.getInstance().registerConcurrencyStrategy(this);
    8. 
  
    
    
  
  8.     }
    9. 
  
    
    
  
    10. 
  
    
    
  
  10.     @Override
    11. 
  
    
    
  
  11.     public <T> Callable<T> wrapCallable(Callable<T> callable) {
    12. 
  
    
    
  
  12.         RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
    13. 
  
    
    
  
  13.         return new WrappedCallable<>(callable, requestAttributes);
    14. 
  
    
    
  
  14.     }
    15. 
  
    
    
  
    16. 
  
    
    
  
  16.     static class WrappedCallable<T> implements Callable<T> {
    17. 
  
    
    
  
    18. 
  
    
    
  
  18.         private final Callable<T> target;
    19. 
  
    
    
  
  19.         private final RequestAttributes requestAttributes;
    20. 
  
    
    
  
    21. 
  
    
    
  
  21.         public WrappedCallable(Callable<T> target, RequestAttributes requestAttributes) {
    22. 
  
    
    
  
  22.             this.target = target;
    23. 
  
    
    
  
  23.             this.requestAttributes = requestAttributes;
    24. 
  
    
    
  
  24.         }
    25. 
  
    
    
  
    26. 
  
    
    
  
  26.         @Override
    27. 
  
    
    
  
  27.         public T call() throws Exception {
    28. 
  
    
    
  
  28.             try {
    29. 
  
    
    
  
  29.                 RequestContextHolder.setRequestAttributes(requestAttributes);
    30. 
  
    
    
  
  30.                 return target.call();
    31. 
  
    
    
  
  31.             } finally {
    32. 
  
    
    
  
  32.                 RequestContextHolder.resetRequestAttributes();
    33. 
  
    
    
  
  33.             }
    34. 
  
    
    
  
  34.         }
    35. 
  
    
    
  
  35.     }
    36. 
  
    
    
  
  36. }
    37.

如代码所示，我们编写了一个 RequestHystrixConcurrencyStrategy ，在其中：

  • wrapCallable 方法拿到 RequestContextHolder.getRequestAttributes() ，也就是我们想传播的对象；

  • 在 WrappedCallable 类中，我们将要传播的对象作为成员变量，并在其中的call方法中，为静态方法设值。

  • 这样，在Hystrix包裹的方法中，就可以使用 RequestContextHolder.getRequestAttributes() 获取到相关属性——也就是说，可以拿到 RequestContextHolder 中的 ThreadLocal 属性。

经过测试，代码能正常工作。

新的问题

至此，我们已经实现了 ThreadLocal 属性的传递，然而Hystrix只允许有一个并发策略！这意味着——如果不做任何处理，Sleuth、Spring Security将无法正常拿到上下文！（上文说过，目前Sleuth、Spring Security都是通过自定义并发策略的方式来传递ThreadLocal对象的。）

如何解决这个问题呢？

我们知道，Spring Cloud中，Spring Cloud Security与Spring Cloud Sleuth是可以共存的！我们不妨参考下Sleuth以及Spring Security的实现：

  • Sleuth： org.springframework.cloud.sleuth.instrument.hystrix.SleuthHystrixConcurrencyStrategy

  • Spring Security： org.springframework.cloud.netflix.hystrix.security.SecurityContextConcurrencyStrategy

阅读完后，你将恍然大悟——于是，我们可以模仿它们的写法，改写上文编写的并发策略：

 
   
   
 
    
  
    
    
  
  1. public class RequestAttributeHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {
    2. 
  
    
    
  
  2.     private static final Log log = LogFactory.getLog(RequestAttributeHystrixConcurrencyStrategy.class);
    3. 
  
    
    
  
    4. 
  
    
    
  
  4.     private HystrixConcurrencyStrategy delegate;
    5. 
  
    
    
  
    6. 
  
    
    
  
  6.     public RequestAttributeHystrixConcurrencyStrategy() {
    7. 
  
    
    
  
  7.         try {
    8. 
  
    
    
  
  8.             this.delegate = HystrixPlugins.getInstance().getConcurrencyStrategy();
    9. 
  
    
    
  
  9.             if (this.delegate instanceof RequestAttributeHystrixConcurrencyStrategy) {
    10. 
  
    
    
  
  10.                 // Welcome to singleton hell...
    11. 
  
    
    
  
  11.                 return;
    12. 
  
    
    
  
  12.             }
    13. 
  
    
    
  
  13.             HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins
    14. 
  
    
    
  
  14.                     .getInstance().getCommandExecutionHook();
    15. 
  
    
    
  
  15.             HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
    16. 
  
    
    
  
  16.                     .getEventNotifier();
    17. 
  
    
    
  
  17.             HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
    18. 
  
    
    
  
  18.                     .getMetricsPublisher();
    19. 
  
    
    
  
  19.             HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
    20. 
  
    
    
  
  20.                     .getPropertiesStrategy();
    21. 
  
    
    
  
  21.             this.logCurrentStateOfHystrixPlugins(eventNotifier, metricsPublisher,
    22. 
  
    
    
  
  22.                     propertiesStrategy);
    23. 
  
    
    
  
  23.             HystrixPlugins.reset();
    24. 
  
    
    
  
  24.             HystrixPlugins.getInstance().registerConcurrencyStrategy(this);
    25. 
  
    
    
  
  25.             HystrixPlugins.getInstance()
    26. 
  
    
    
  
  26.                     .registerCommandExecutionHook(commandExecutionHook);
    27. 
  
    
    
  
  27.             HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
    28. 
  
    
    
  
  28.             HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
    29. 
  
    
    
  
  29.             HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
    30. 
  
    
    
  
  30.         }
    31. 
  
    
    
  
  31.         catch (Exception e) {
    32. 
  
    
    
  
  32.             log.error("Failed to register Sleuth Hystrix Concurrency Strategy", e);
    33. 
  
    
    
  
  33.         }
    34. 
  
    
    
  
  34.     }
    35. 
  
    
    
  
    36. 
  
    
    
  
  36.     private void logCurrentStateOfHystrixPlugins(HystrixEventNotifier eventNotifier,
    37. 
  
    
    
  
  37.             HystrixMetricsPublisher metricsPublisher,
    38. 
  
    
    
  
  38.             HystrixPropertiesStrategy propertiesStrategy) {
    39. 
  
    
    
  
  39.         if (log.isDebugEnabled()) {
    40. 
  
    
    
  
  40.             log.debug("Current Hystrix plugins configuration is ["
    41. 
  
    
    
  
  41.                     + "concurrencyStrategy [" + this.delegate + "]," + "eventNotifier ["
    42. 
  
    
    
  
  42.                     + eventNotifier + "]," + "metricPublisher [" + metricsPublisher + "],"
    43. 
  
    
    
  
  43.                     + "propertiesStrategy [" + propertiesStrategy + "]," + "]");
    44. 
  
    
    
  
  44.             log.debug("Registering Sleuth Hystrix Concurrency Strategy.");
    45. 
  
    
    
  
  45.         }
    46. 
  
    
    
  
  46.     }
    47. 
  
    
    
  
    48. 
  
    
    
  
  48.     @Override
    49. 
  
    
    
  
  49.     public <T> Callable<T> wrapCallable(Callable<T> callable) {
    50. 
  
    
    
  
  50.         RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
    51. 
  
    
    
  
  51.         return new WrappedCallable<>(callable, requestAttributes);
    52. 
  
    
    
  
  52.     }
    53. 
  
    
    
  
    54. 
  
    
    
  
  54.     @Override
    55. 
  
    
    
  
  55.     public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
    56. 
  
    
    
  
  56.             HystrixProperty<Integer> corePoolSize,
    57. 
  
    
    
  
  57.             HystrixProperty<Integer> maximumPoolSize,
    58. 
  
    
    
  
  58.             HystrixProperty<Integer> keepAliveTime, TimeUnit unit,
    59. 
  
    
    
  
  59.             BlockingQueue<Runnable> workQueue) {
    60. 
  
    
    
  
  60.         return this.delegate.getThreadPool(threadPoolKey, corePoolSize, maximumPoolSize,
    61. 
  
    
    
  
  61.                 keepAliveTime, unit, workQueue);
    62. 
  
    
    
  
  62.     }
    63. 
  
    
    
  
    64. 
  
    
    
  
  64.     @Override
    65. 
  
    
    
  
  65.     public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
    66. 
  
    
    
  
  66.             HystrixThreadPoolProperties threadPoolProperties) {
    67. 
  
    
    
  
  67.         return this.delegate.getThreadPool(threadPoolKey, threadPoolProperties);
    68. 
  
    
    
  
  68.     }
    69. 
  
    
    
  
    70. 
  
    
    
  
  70.     @Override
    71. 
  
    
    
  
  71.     public BlockingQueue<Runnable> getBlockingQueue(int maxQueueSize) {
    72. 
  
    
    
  
  72.         return this.delegate.getBlockingQueue(maxQueueSize);
    73. 
  
    
    
  
  73.     }
    74. 
  
    
    
  
    75. 
  
    
    
  
  75.     @Override
    76. 
  
    
    
  
  76.     public <T> HystrixRequestVariable<T> getRequestVariable(
    77. 
  
    
    
  
  77.             HystrixRequestVariableLifecycle<T> rv) {
    78. 
  
    
    
  
  78.         return this.delegate.getRequestVariable(rv);
    79. 
  
    
    
  
  79.     }
    80. 
  
    
    
  
    81. 
  
    
    
  
  81.     static class WrappedCallable<T> implements Callable<T> {
    82. 
  
    
    
  
    83. 
  
    
    
  
  83.         private final Callable<T> target;
    84. 
  
    
    
  
  84.         private final RequestAttributes requestAttributes;
    85. 
  
    
    
  
    86. 
  
    
    
  
  86.         public WrappedCallable(Callable<T> target, RequestAttributes requestAttributes) {
    87. 
  
    
    
  
  87.             this.target = target;
    88. 
  
    
    
  
  88.             this.requestAttributes = requestAttributes;
    89. 
  
    
    
  
  89.         }
    90. 
  
    
    
  
    91. 
  
    
    
  
  91.         @Override
    92. 
  
    
    
  
  92.         public T call() throws Exception {
    93. 
  
    
    
  
  93.             try {
    94. 
  
    
    
  
  94.                 RequestContextHolder.setRequestAttributes(requestAttributes);
    95. 
  
    
    
  
  95.                 return target.call();
    96. 
  
    
    
  
  96.             }
    97. 
  
    
    
  
  97.             finally {
    98. 
  
    
    
  
  98.                 RequestContextHolder.resetRequestAttributes();
    99. 
  
    
    
  
  99.             }
    100. 
  
    
    
  
  100.         }
    101. 
  
    
    
  
  101.     }
    102. 
  
    
    
  
  102. }
    103.

简单讲解下：

  • 将现有的并发策略作为新并发策略的成员变量

  • 在新并发策略中，返回现有并发策略的线程池、Queue。

Pull Request

笔者已将该实现方式Pull Request：https://github.com/spring-cloud/spring-cloud-netflix/pull/2509 ，希望官方能够接纳，也希望在不久的将来，能够更舒服、更爽地使用Spring Cloud。

PS. Pull Request的代码跟博客中的代码略有区别，有少量简单的优化，主要是增加了一个开关。

灵感来自

  • https://stackoverflow.com/questions/34062900/forward-a-request-header-with-a-feign-client-requestinterceptor

  • https://stackoverflow.com/questions/34719809/unreachable-security-context-using-feign-requestinterceptor

  • https://github.com/spring-cloud/spring-cloud-netflix/pull/1379

  • https://github.com/Writtscher/kanoon


以上是关于实用技巧：Hystrix传播ThreadLocal对象（两种方案）的主要内容，如果未能解决你的问题，请参考以下文章

hystrix源码之hystrix请求变量

SpringCloud断路器Hystrix全面解析

生产技巧：Feign如何控制Hystrix的启停超时熔断？

微服务架构Spring Cloud之Hystrix

8-29 实战技巧：如何设置线程池

全方位，多角度理解ThreadLocal（强烈推荐）