（给ImportNew加星标，提高Java技能）

[本文来自：www.ii77.com]

转自：崔世宁，

http://www.cnblogs.com/kingszelda/p/10312242.html

1. 什么是Hystrix

Hystrix是Netflix的一个开源框架，地址如下：http://github.com/Netflix/Hystrix

中文名为“豪猪”，即平时很温顺，在感受到危险的时候，用刺保护自己；在危险过去后，还是一个温顺的肉球。

所以，整个框架的核心业务也就是这2点：

1. 何时需要保护

2. 如何保护

2. 何时需要保护

对于一个系统而言，它往往承担着2层角色，服务提供者与服务消费者。对于服务消费者而言最大的痛苦就是如何“明哲保身”，做过网关项目的同学肯定感同身受

上面是一个常见的系统依赖关系，底层的依赖往往很多，通信协议包括 socket、HTTP、Dubbo、WebService等等。当通信层发生网络抖动以及所依赖的系统发生业务响应异常时，我们业务本身所提供的服务能力也直接会受到影响。

这种效果传递下去就很有可能造成雪崩效应，即整个业务联调发生异常，比如业务整体超时，或者订单数据不一致。

那么核心问题就来了，如何检测业务处于异常状态？

成功率！成功率直接反映了业务的数据流转状态，是最直接的业务表现。

当然，也可以根据超时时间做判断，比如 Sentinel 的实现。其实这里概念上可以做一个转化，用时间做超时控制，超时=失败，这依然是一个成功率的概念。

3. 如何保护　

如同豪猪一样，“刺”就是他的保护工具，所有的攻击都会被刺无情的怼回去。

在 Hystrix 的实现中，这就出现了“熔断器”的概念，即当前的系统是否处于需要保护的状态。

当熔断器处于开启的状态时，所有的请求都不会真正的走之前的业务逻辑，而是直接返回一个约定的信息，即 FallBack。通过这种快速失败原则保护我们的系统。 

但是，系统不应该永远处于“有刺”的状态，当危险过后需要恢复正常。

于是对熔断器的核心操作就是如下几个功能：

1. 如果成功率过低，就打开熔断器，阻止正常业务

2. 随着时间的流动，熔断器处于半打开状态，尝试性放入一笔请求

　　熔断器的核心 API 如下图： 

4. 限流、熔断、隔离、降级

这四个概念是我们谈起微服务会经常谈到的概念，这里我们讨论的是 Hystrix 的实现方式。

限流

• 这里的限流与 Guava 的 RateLimiter 的限流差异比较大，一个是为了“保护自我”，一个是“保护下游”

• 当对服务进行限流时，超过的流量将直接 Fallback，即熔断。而 RateLimiter 关心的其实是“流量整形”，将不规整流量在一定速度内规整

熔断

• 当我的应用无法提供服务时，我要对上游请求熔断，避免上游把我压垮

• 当我的下游依赖成功率过低时，我要对下游请求熔断，避免下游把我拖垮

降级

• 降级与熔断紧密相关，熔断后业务如何表现，约定一个快速失败的 Fallback，即为服务降级

隔离

• 业务之间不可互相影响，不同业务需要有独立的运行空间

• 最彻底的，可以采用物理隔离，不同的机器部

• 次之，采用进程隔离，一个机器多个 Tomcat

• 次之，请求隔离

• 由于 Hystrix 框架所属的层级为代码层，所以实现的是请求隔离，线程池或信号量

5. 源码分析

先上一个 Hystrix 的业务流程图

可以看到 Hystrix 的请求都要经过 HystrixCommand 的包装，其核心逻辑在 AbstractComman.java 类中。

下面的源码是基于 RxJava 的，看之前最好先了解下 RxJava 的常见用法与逻辑，否则看起来会很迷惑。

简单的说，RxJava 就是基于回调的函数式编程。通俗的说，就等同于策略模式的匿名内部类实现。

5.1 熔断器

首先看信号量是如何影响我们请求的：

private Observable applyHystrixSemantics(final AbstractCommand _cmd) {
        // 自定义扩展
        executionHook.onStart(_cmd);

        //判断熔断器是否允许请求过来
        if (circuitBreaker.attemptExecution()) {
　　　　　　　//获得分组信号量，如果没有采用信号量分组，返回默认通过的信号量实现
            final TryableSemaphore executionSemaphore = getExecutionSemaphore();
            final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
　　　　　　　//调用终止的回调函数
            final Action0 singleSemaphoreRelease = new Action0() {
                @Override
                public void call() {
                    if (semaphoreHasBeenReleased.compareAndSet(false, true)) {
                        executionSemaphore.release();
                    }
                }
            };
　　　　　　　//调用异常的回调函数
            final Action1 markExceptionThrown = new Action1() {
                @Override
                public void call(Throwable t) {
                    eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey);
                }
            };
　　　　　　　//根据信号量尝试竞争信号量
            if (executionSemaphore.tryAcquire()) {
                try {
                    //竞争成功，..执行参数
                    executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
                    return executeCommandAndObserve(_cmd)
                            .doOnError(markExceptionThrown)
                            .doOnTerminate(singleSemaphoreRelease)
                            .doOnUnsubscribe(singleSemaphoreRelease);
                } catch (RuntimeException e) {
                    return Observable.error(e);
                }
            } else {
　　　　　　　　　　//竞争失败，进入fallback
                return handleSemaphoreRejectionViaFallback();
            }
        } else {
　　　　　　　 //熔断器已打开，进入fallback
            return handleShortCircuitViaFallback();
        }
    }

什么时候熔断器可以放请求进来：