Hystrix系列之执行原理分析

Posted 2021-04-30 占小狼的博客

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Hystrix系列之执行原理分析相关的知识，希望对你有一定的参考价值。

上文《》介绍了Hystrix的由来，本文会深入分析Hystrix的执行过程。

Hystrix的大部分逻辑基于RxJava进行实现，让很热多人望而却步，停留在了仅仅使用的地步，今天从一个简单的HelloWorld开始。

 
   
   
 
  
    
    
  public class CommandHelloWorld extends HystrixCommand<String> {
  
    
    
  
  
    
    
      private final String name;
  
    
    
  
  
    
    
      public CommandHelloWorld(String name) {
  
    
    
          super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
  
    
    
          this.name = name;
  
    
    
      }
  
    
    
  
  
    
    
      @Override
  
    
    
      protected String run() {
  
    
    
          // 省略业务逻辑
  
    
    
          // 该方法可能会抛出异常
  
    
    
          return "Hello " + name + "!";
  
    
    
      }
  
    
    
  
  
    
    
      @Override
  
    
    
      protected String getFallback() {
  
    
    
          return "Hello Failure " + name + "!";
  
    
    
      }
  
    
    
  }

通过简单的实现run方法和getFallback 方法， CommandHelloWorld具备了熔断降级的功能，其中 HystrixCommand提供了4个方法：execute()，queue()，observe()，toObservable()，平时只需要关注execute和queue即可。

queue()：异步调用，返回一个Future对象，后面可以通过Future获取结果。
execute() 同步调用，调用后直接block住，直到依赖服务返回结果，或者抛出异常。

 
   
   
 
  
    
    
  public R execute() {
  
    
    
      try {
  
    
    
          return queue().get();
  
    
    
      } catch (Exception e) {
  
    
    
          throw Exceptions.sneakyThrow(decomposeException(e));
  
    
    
      }
  
    
    
  }

其实同步方式是通过直接执行Future的get方法进行实现的，这里需要知道这个返回的Future对象到底是什么？

通过实现可以发现，返回的Futrue对象只是对 toObservable返回结果的封装代理，把注意力转移到 toObservable方法的实现。

第一次看到这个方法的实现，也许会很崩溃，方法内部初始化了一系列的Action0对象，这是RxJava的内部对象，可以看作是订阅一个事件后的回调。

一开始，不要太过在意这些Action，不然会陷入迷失，在debug的时候，通过不断的回调，经常会不知身处何处。

直接看到 toObservable方法的return处，又是一坨回调，崩溃。

Hystrix系列之执行原理分析

如果不熟悉RxJava语法，看这种代码真心的累，call方法的前面部分主要做两件事。 1、记录请求日志（日志功能开启） 2、从缓存中返回结果（定义了cacheKey方法，并且缓存功能开启）

如果缓存没有开启，或者返回null，那只能执行正常逻辑从下游服务拿取数据，这里通过上面定义的 applyHystrixSemantics Action进行回调，最终执行的是 applyHystrixSemantics方法，这个方法才是精华所在，想调试的同学，直接在这个方法入口打个断点，事半功倍。

Hystrix系列之执行原理分析

其中 circuitBreaker.attemptExecution()的返回结果，决定了接下去是执行正常逻辑、还是降级逻辑，这才是精华的精华。

Hystrix系列之执行原理分析

看一下熔断器的 attemptExecution方法，内部涉及了多个开关

forceOpen 强制开启，所以请求都执行降级逻辑
forceClose 强制关闭，所以请求都执行正常逻辑
circuitOpened 熔断开关，默认为-1，请求执行正常逻辑，如果发生熔断，该值会被修改成0，请求执行降级逻辑
HALF_OPEN 熔断半开，即熔断之后，每隔一段会进行试探

个人觉得，这里的实现过于复杂。

如果没有发生熔断，还有一道门槛，Hystrix提供了一个信号量限流器，限制进入熔断器最大并发数，可以控制请求下游的并发量，如果超过这个阈值，会被降级处理，有效的保护下游服务不会被突发流量给攻击。

通过的请求，继续调用 executeCommandAndObserve方法，在该方法中，又定义了一堆让人迷惑的Action，不过这次通过名字和实现，可以知道个大概，先把这些Action放一边，看看接下去会执行的方法。

 
   
   
 
  
    
    
  private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
  
    
    
       // 忽略一堆Action的定义
  
    
    
      Observable<R> execution;
  
    
    
      if (properties.executionTimeoutEnabled().get()) {
  
    
    
          execution = executeCommandWithSpecifiedIsolation(_cmd)
  
    
    
                  .lift(new HystrixObservableTimeoutOperator<R>(_cmd));
  
    
    
      } else {
  
    
    
          execution = executeCommandWithSpecifiedIsolation(_cmd);
  
    
    
      }
  
    
    
      return execution.doOnNext(markEmits)
  
    
    
          .doOnCompleted(markOnCompleted)
  
    
    
          .onErrorResumeNext(handleFallback)
  
    
    
          .doOnEach(setRequestContext);
  
    
    
  }

Hystrix内部提供了超时检查的机制，如果参数 executionTimeoutEnabled开启，则每次请求都会提交一个任务到线程池中延迟执行。由于Hystrix实现中考虑的东西太多，所以在实现上还是很复杂。

Hystrix系列之执行原理分析

这里只给出了关键逻辑，如果配置了超时时间10ms，会提交一个延迟10ms执行的任务，其中 tick方法会通过CAS机制保证超时状态的变更，最终对应command的会执行onError方法，这里加入的 HystrixContextRunnable主要为了跨线程的上下文数据传递。

在执行正常逻辑的实现中，Hystrix内部提供了信号量、线程池两种模式，默认使用线程池模式。在 executeCommandWithSpecifiedIsolation方法中，分别对这两种模式进行了处理，而且可以根据参数随时进行切换，这就是为什么线程池一开始就要初始化的原因，虽然有资源的消耗，但是带来了更好的灵活性，在需要的时候可以从信号量模式变成线程池模式进行隔离。

下面以信号量的方式为例，分析下如何执行用户自定义的 run方法.

Hystrix系列之执行原理分析

又想吐槽Hystrix的代码，又是这种回调，在call实现中，暂时忽略前面的一大坨逻辑，跟进 getUserExecutionObservable方法。

Hystrix系列之执行原理分析

继续查看 getExecutionObservable方法，该方法在 HystrixCommand中被重写实现。

run方法终于执行了。 run方法执行之后，如果正常返回、抛出异常、或者其它情况，都需要对应的后续处理，这时之前 executeCommandAndObserve方法中定义的Action，就开始起作用了。

 
   
   
 
  
    
    
  execution.doOnNext(markEmits)
  
    
    
      .doOnCompleted(markOnCompleted)
  
    
    
      .onErrorResumeNext(handleFallback)
  
    
    
      .doOnEach(setRequestContext);