Hystrix

5b9bd19a91262

简介

Hystrix是NetFlix公司推出的用于微服务软件架构设计的服务保护模块，该模块是目的是为了去解决整体服务其中某一个服务出现宕机而依赖该服务的其余节点进入故障不可用状态出现调用雪崩这类问题。对微服务架构架构的延迟和故障去提供一个很好的容错性，这个模块提供了诸如服务降级，熔断，线程隔离，请求合并以及监控。

源码分析

Hystrix的Work flow

说到熔断呢，这就要说到熔断器模式了，这个熔断器最早呢就是在电路中，保护线路过载的，当某一处发生短路之后，这个断路器能够及时的打开，避免线路内的电器过载导致用出现故障，起火甚至爆炸。在微服务模式下呢，(说实话是真不喜欢用这个词，TM现在是个公司面试就问你用没用过Spring Cloud，Dobbo 微服务之类的，你TM日访问量都不足1000，要什么微服务，要啥自行车。MVC就不能支撑你了嘛),在微服务架构下(真香.jpg)，断路器也是这个作用。我们直接看Hystrix Circuit-Breaker

5b9bd845a317a

上图就是Hystrix的Work flow,我们跟着官网的流程图走，使用我的天魔肢解大法来跟着流程图的序号一步一步的揭开Hystrix Circuit-Breaker 的面纱

Hystrix是基于Rxjava的，所以阅读Hystrix源码需要了解Rxjava的机制 Rxjava

先看第一步是需要创建一个HystrixCommand对象或者HystrixObservableCommand对象,看到这两个类名就知道这是设计模式的命令模式，所以先小提一下命令模式和Rxjava

命令模式

将处理方式以命令的形式包裹在对象中，调用方通过调用不同的对象去执行不同的命令的。命令模式的核心就是定义三个类：1，命令执行对象 2、命令接口 3、命令的调用方对象

Rxjava

Observable:发射源，被观察者

Observer:接收源，观察者

Subscriber:订阅者，接收源

Subject:Subject既可充当发射源，也可充当接收源

Subscriber跟Observer之间的区别在于Subscriber实现了Observer接口，比Observer多了一个最重要的方法unsubscribe( )，用来取消订阅，当你不再想接收数据了，可以调用unsubscribe( )方法停止接收，还多了一个方法是onStart()，它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。另外Observer 在 subscribe() 过程中,最终也会被转换成 Subscriber 对象。

 Observable.create(new Observable.OnSubscribe() {
            @Override
            public void call(Subscriber subscriber) {
                subscriber.onNext("Rxjava 绝壁不好用");
                subscriber.onNext("我王境泽就是饿死都不会去学Rxjava");
                subscriber.onNext("Rxjava 真香");
                subscriber.onCompleted();
            }
        }).subscribe(new Subscriber() {
            @Override
            public void onCompleted() {

            }
            @Override
            public void onError(Throwable throwable) {

            }
            @Override
            public void onNext(Object o) {
                System.out.println("Subscribe Message"+o);
            }
        });

在这个简单的示例中，我展示了RxJava的观察订阅模式(cold Observable观察者)，使用Observable.create去创建一个被观察者，使用new Subscriber来创建一个订阅方，二者通过被观察者的订阅方法.subscribe来建立观察订阅关系，在这里你可能会感觉到这个方法调用略显拗口，为何是被观察者订阅订阅方，这里其实是为了更好的满足更加便捷的调用链，故而如此设计。Observable对象包括Hot Observable,cold Observable. 对于第一种来说，当被观测对象创建的时候就回去发射事件，所以该模式下额订阅可能拿到的数据只是局部数据，另外一种呢就是直到订阅者与之建立关系之后才会去发射时间，它会哪里阻塞等待。可以使用被观察者对象调用publish方法将cold Observable 转换为 Hot Observable，此外还可以通过 Subject或者Processor将cold Observable 转换为 Hot Observable,Processor 是 RxJava2.x 新增的类，继承自 Flowable 支持背压控制。而 Subject 则不支持背压控制，同时Subject和publish方法不一致的是，Subject不是线程安全的。

命令的执行流程

HystrixCommandHystrixObservableCommand 分别实现了AbstractCommand中的execute,queue和observe,toObservable命令，其区别在于execute是同步的其执行方式为调用queue().get()方法，而queue是异步方法返回一个Future对象。观察HystrixCommand#queue，可以看到核心语句为调用HystrixCommand的toObservable()方法。

 final Future delegate = toObservable().toBlocking().toFuture();

而HystrixObservableCommand则直接调用的是observe,toObservable。observe也是核心也是通过调用toObservable方法，只不过是observe方法用Subject对象将toObservable返回的cold Observable对象转换为了Hot Observable。纵观这几个方法其核心调用都在于toObservable方法，我么接下来就直接查看toObservable方法

public Observable toObservable() {
        final AbstractCommand _cmd = this;
        final Action0 terminateCommandCleanup ...
        final Action0 unsubscribeCommandCleanup ...
        final Func0> applyHystrixSemantics ...
        final Func1 wrapWithAllOnNextHooks ...
        final Action0 fireOnCompletedHook ...
        return Observable.defer(new Func0>() {
            @Override
            public Observable call() {
                if (!commandState.compareAndSet(CommandState.NOT_STARTED, CommandState.OBSERVABLE_CHAIN_CREATED)) {
                    IllegalStateException ex = new IllegalStateException("This instance can only be executed once. Please instantiate a new instance.");
                    throw new HystrixRuntimeException(FailureType.BAD_REQUEST_EXCEPTION, _cmd.getClass(), getLogMessagePrefix() + " command executed multiple times - this is not permitted.", ex, null);
                }
                commandStartTimestamp = System.currentTimeMillis();
                if (properties.requestLogEnabled().get()) {
                    if (currentRequestLog != null) {
                        currentRequestLog.addExecutedCommand(_cmd);
                    }
                }
                final boolean requestCacheEnabled = isRequestCachingEnabled();
                final String cacheKey = getCacheKey();
                if (requestCacheEnabled) {
                    HystrixCommandResponseFromCache fromCache = (HystrixCommandResponseFromCache) requestCache.get(cacheKey);
                    if (fromCache != null) {
                        isResponseFromCache = true;
                        return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                    }
                }
                Observable hystrixObservable =Observable.defer(applyHystrixSemantics).map(wrapWithAllOnNextHooks);
                Observable afterCache;
                if (requestCacheEnabled && cacheKey != null) {
                    HystrixCachedObservable toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
                    HystrixCommandResponseFromCache fromCache = (HystrixCommandResponseFromCache) requestCache.putIfAbsent(cacheKey, toCache);
                    if (fromCache != null) {
                        toCache.unsubscribe();
                        isResponseFromCache = true;
                        return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                    } else {
                        afterCache = toCache.toObservable();
                    }
                } else {
                    afterCache = hystrixObservable;
                }
                return afterCache
                        .doOnTerminate(terminateCommandCleanup)   
                        .doOnUnsubscribe(unsubscribeCommandCleanup) 
                        .doOnCompleted(fireOnCompletedHook);
            }
        });
    }

直接看Observable.defer 这个defer中的代码直到被订阅才会执行，这也是不同于create方法的点，同时defer方法也不需要在call中手动调用onCompleted方法。Action0,Func0 是Rxjava中的接口。Action系列其中为无返回(参数从0到9)call方法，在调用是会被执行，至于Func0系列则是有返回值(参数从0到9)call方法。了解了这些概念，我们再来看toObservable思路就明确了很多,首先看call方法内部，开始处是一个校验一个命令只能被执行一次，然后记录命令的开始执行时间戳，用于之后的判定超时时使用，然后就记录日志，判断是否启用命令缓存，如果开启使用的话并且缓存key不为空，直接从通过中缓存的HystrixCachedObservable经过handleRequestCacheHitAndEmitValues 方法获取一个Observable 对象，接着往下看语句Observable hystrixObservable =Observable.defer(applyHystrixSemantics).map(wrapWithAllOnNextHooks); 该语句是获取一个执行命令的Observable 对象, 接着是如果开启了缓存并且命中的话，去更新缓存中的HystrixCachedObservable 对象, 如果不存在的话，则解除订阅，这样可以避免没有缓冲时在多线程访问下，有且仅有一个线程订阅执行命令，当缓存put成功之后接下来就是通过HystrixCachedObservable#toObservable()取缓存中的cachedObservable ,如果说没开启缓存的话直接取hystrixObservable ，再返回之前做一些预处理，当执行{@code onCompleted} 和{@code onError} 之前做一步清理, 清理逻辑见fireOnCompletedHook terminateCommandCleanup unsubscribeCommandCleanup 。

handleRequestCacheHitAndEmitValues

private Observable handleRequestCacheHitAndEmitValues(final HystrixCommandResponseFromCache fromCache, final AbstractCommand _cmd) { 
        try {
            executionHook.onCacheHit(this);
        } catch (Throwable hookEx) {
            logger.warn("Error calling HystrixCommandExecutionHook.onCacheHit", hookEx);
        }
        return fromCache.toObservableWithStateCopiedInto(this)
                .doOnTerminate(new Action0() {
                    @Override
                    public void call() {
                        if (commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.TERMINAL)) {
                            cleanUpAfterResponseFromCache(false); //user code never ran
                        } else if (commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.TERMINAL)) {
                            cleanUpAfterResponseFromCache(true); //user code did run
                        }
                    }
                })
                .doOnUnsubscribe(new Action0() {
                    @Override
                    public void call() {
                        if (commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
                            cleanUpAfterResponseFromCache(false); //user code never ran
                        } else if (commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.UNSUBSCRIBED)) {
                            cleanUpAfterResponseFromCache(true); //user code did run
                        }
                    }
                });
    }

Action#terminateCommandCleanup

 final Action0 terminateCommandCleanup = new Action0() {
            @Override
            public void call() {
                if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.TERMINAL)) {
                    handleCommandEnd(false);
                } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.TERMINAL)) {
                    handleCommandEnd(true); 
                }
            }
        };

Action0#terminateCommandCleanup会把命令的commandState设置为TERMINAL。当调用toObservable方法时候会先去先去校验当前命令的执行状态，如果状态为CommandState.OBSERVABLE_CHAIN_CREATED则证明上一个命令从来没有执行过只是创建了Obsservable，则将其标记为命令终止，开始下一次命令执行序列，如果说内存中命令状态为USER_CODE_EXECUTED，则证明其命令的确执行了，同时也要将其设置为终止，进而接受订阅。

Action#unsubscribeCommandCleanup

        final Action0 unsubscribeCommandCleanup = new Action0() {
            @Override
            public void call() {
                if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
                    if (!_cmd.executionResult.containsTerminalEvent()) {
                        _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
                        try {
                            executionHook.onUnsubscribe(_cmd);
                        } catch (Throwable hookEx) {
                            logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
                        }
                        _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
                                .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
                    }
                    handleCommandEnd(false); //user code never ran
                } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.UNSUBSCRIBED)) {
                    if (!_cmd.executionResult.containsTerminalEvent()) {
                        _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
                        try {
                            executionHook.onUnsubscribe(_cmd);
                        } catch (Throwable hookEx) {
                            logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
                        }
                        _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
                                .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
                    }
                    handleCommandEnd(true); //user code did run
                }
            }
        };

Action#unsubscribeCommandCleanup 根据命令有无执行标记，统一将命令的状态设置为UNSUBSCRIBED状态，当被标记为该状态是不执行命令的，判断该命令的executionResult是否有终止事件，如果没有则需要标记该executionResult的event的状态为取消,执行handleCommandEnd方法，主要是调用metrics.markCommandDone去对 executionResult做处理，executionResult是Hystrix的命令执行结果的记录，包含生命周期内的全部信息，调用metrics.markCommandDone实际上调用的是HystrixThreadEventStream.getInstance().executionDone

private void handleCommandEnd(boolean commandExecutionStarted) {
       //无关代码...
        executionResult = executionResult.markUserThreadCompletion((int) userThreadLatency);
 			//省略分支
            metrics.markCommandDone(executionResultAtTimeOfCancellation, commandKey, threadPoolKey, commandExecutionStarted);
        }
			//无关代码...
    }


public void executionDone(ExecutionResult executionResult, HystrixCommandKey commandKey, HystrixThreadPoolKey threadPoolKey) {
        HystrixCommandCompletion event = HystrixCommandCompletion.from(executionResult, commandKey, threadPoolKey);
        writeOnlyCommandCompletionSubject.onNext(event);
    }

看上述代码，则发现根据入参封装一个HystrixCommandCompletion event，这个Event是封装了executionResult和requestContext，在这个状态下，他保存的是命令执行完成的完整信息，调用writeOnlyCommandCompletionSubject.onNext(event);实际上调用的是writeOnlyCommandCompletionSubject的writeCommandCompletionsToShardedStreams()方法将该命令的信息写入，用于之后的指令汇聚，统计。

熔断Hystrix Circuit-Breaker

5bacd439d9c7e

没错就是这个，你看到的上图就是我要说的断路器。。。，常见的电路保护装置，当然在我们的软件设计中同样也是有这样的设计用来防止雪崩，这就是Hystrix的Circuit Breaker，下图呢就是上面的HystrixCommand``HystrixObservableCommand 和Circuit Breaker的交互逻辑

5bacd552aab66

正常状态下，这个我们都知道，可以正常访问所依赖的远端服务，得到结果

当一段时间内，请求的失败率大于我们配置的阈值的时候，断路器就会打开，当Circuit Breaker打开的时候是不会访问远端服务的，而是会走短路Fallback，请求本地的降级服务，

当这个断路器打开一段时间之后会变为半开状态，此时用于试探的访问上一次成功的服务，如果成功的话会关闭断路器，失败的话会继续变为打开状态。

上述呢就是断路器的运行流程，接着我们来看一下Netflix公司对该模式的具体设计先看断路器的类图设计

3534

一共只有2个实现，分别为HystrixCircuitBreakerImpl和 NoOpCircuitBreaker, NoOpCircuitBreaker 为配置项断路器关闭时，默认关闭断路器，该类的实现简单的可怕，所有的实现方法都返回true;接着看下断路器中的方法

public interface HystrixCircuitBreaker {
    boolean allowRequest();
    boolean isOpen();
    void markSuccess();
    void markNonSuccess();
    boolean attemptExecution();
}

HystrixCircuitBreakerImpl中该接口的实现方法呢则就是Hystrix的断路器设计的全部精华了，

allowRequest方法呢就是的去读取断路器的状态，他不会去修改断路器的状态所以是幂等操作，而attemptExecution除了上述的读取断路器状态，还会去修改断路器的状态(判断的条件为如果当前时间已经超过了配置项中设置的断路器的关闭持续时间，则会尝试将开启的断路器变为半开启).

isOpen方法就是单纯的判断断路器的状态，除了判断配置项，还会读取内存中的circuitOpened的值是否大于0，来判断断路器的状态，断路器关闭状态该值为-1，当断路器处于开启状态，该值为断路器开启当时的时间戳，该值同时也用作attemptExecution去转换开启状态为半开启状态的持续时间做比较运算。

markSuccess方法呢就是会执行一波cas操作，将原子变量status由半开转为关闭，如果成功了之后就重置指标收集的流，设置新的订阅者，最后设置内存中的circuitOpened的值为-1。

markNonSuccess呢就是将原子变量status由半开状态转为开启状态，同是设置内存中的circuitOpened的值为当时的时间戳。

结合上图的断路器的官网设计图和HystrixCircuitBreakerImpl的实现方法，我呢就介绍了整个断路器的工作原理，具体的命令执行和断路器实在那个位置相互关联起来呢。这个是个关键，我们观察AbstractCommand#applyHystrixSemantics方法就明白了。

Fallback

我们接着来看回退逻辑依旧在AbstractCommand#applyHystrixSemantics方法中else分支走的方法handleShortCircuitViaFallback,回退逻辑呢一般是由开发人员自己编写的，存放在commandContainsFallback这个ConcurrentHashMap中，HystrixObservableCommand找类中的resumeWithFallback方法，而HystrixCommand则是找类中的getFallback方法去执行

private Observable handleShortCircuitViaFallback() {
        eventNotifier.markEvent(HystrixEventType.SHORT_CIRCUITED, commandKey);
        Exception shortCircuitException = new RuntimeException("Hystrix circuit short-circuited and is OPEN");
        executionResult = executionResult.setExecutionException(shortCircuitException);
        try {
            return getFallbackOrThrowException(this, HystrixEventType.SHORT_CIRCUITED, FailureType.SHORTCIRCUIT,
                    "short-circuited", shortCircuitException);
        } catch (Exception e) {
            return Observable.error(e);
        }
    }

实战使用

在Spring Cloud中使用Hystrix非常简单，只需要在pom文件中引入对应的依赖，或者是Gradle的话加入对应的compile，然后启动类的上方增加@EnableCircuitBreaker或者是@EnableHystrix 即可将Hystrix整合入项目中，

@EnableCircuitBreaker
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
}

我们引入Hystrix的目的是为了防止雪崩，让上下游服务的调用不应为下游服务的突然宕机，导致整个系统工程的瘫痪，难以访问，具体入项目中呢就是对远端方法的容错性。

@RestController
public class DemoController {
  @AutoWired
  ResttTemplete restTemplete; 
    @RequestMapping("/queryRemote")
    @HystrixCommand(fallbackMethod ="fallbackMethod" )
    public Map queryRemoteService(){
      Map result=restTemplete.getForObject("http://XXXXXXX",Map.class);   
      return result;
    }
    public Map fallbackMethod(){
         Map result=new HashMap<>();
      	 return result;
    }
}

上述代码使用了@HystrixCommand(fallbackMethod ="FallbackMethod" )同时还增加了一个FallbackMethod的回退方法，学习过上述源码解读的同学可以明白，当远端服务宕机一段时间内的调用错误率达到阈值，那么我们的断路器会打开，Hystrix一共有两种方式来处理，1抛出调用异常(默认),2找寻该调用方法的回退逻辑，该注解中的fallbackMethod中的值就是其对应的回退，但是当请求失败，被拒绝，超时都会进入回退方法，要实际的了解到断路器是否打开则需要对断路器的状态进行健康检查。需要引入actuator，断路器的状态在/health返回的结果中可以展示。

上面是单独使用Hystrix的演示，在和Feign整合，之后可以使用回退类来对整个远端调用接口的降级处理，回退类实现远端调用接口的全部方法，然后就不用在单个使用@HystrixCommand()的类中写回退方法，这样方便统一管理，和后期维护。

最后一个完整的项目，必不可少的是日志环节，应为随着各个服务的独立，系统被拉伸，整体的规模变得非常的庞大，各个服务之间的调用，虽有了断路器用于保障，不至于系统瘫痪，但是在回退方法的处理要增加日志的处理，则使用@FeignClient的配置fallbackFactory来处理回退，可以拿到调用失败的原因，存入日志体系，用于快速响应和日志框架的分析。

Hystrix指标监控和可视化

Hystrix对于监控可以说是实时的，在浏览源码的过程我们知道，有几个stream的存在用于收集metric指标，每分钟请求数，请求成功和失败的个数，如错误率，对应调用的方法信息，等等，访问ip:port/hystrix.stream即可打印出data节点。

Hystrix的数据可视化依托于Dashboard插件，同样在引入该插件之后，在启动类上增加 @EnableHystrixDashboard 启动服务访问ip:port/hystrix即可查看

上述的查看方式都是针对单一实例的，在一个大型项目呢介于Ribbon的存在，每个服务都是多实例的情况，那这样的话，想看每个实例的指标情况就得每个实例上都增加该插件，着实麻烦，针对这些情况我们可以引入Turbine来聚合监控的这个metric信息。可以单独把Turbine整合进入原项目，这样很方便即可拿到服务实时调用的情况，如笔者公司的系统，监控指标可视化仅在公司内网使用，不暴露出去，处理的方式是使用Turbine-stream插件放如消息队列中，有Turbine去获取消息队列的数据进行可视化展示，而不是直连线上服务。

ok本次的Hystrix的开篇就先到这里，至于Hystrix的隔离策略在之后继续补充。end 2018-10-15 22:47:24