SpringCloud容错处理：Hystrix源码分析

左手的ㄟ右手 2023-07-02 15:23 8阅读 0赞

# 容错处理 #

> 容错处理是指软件运行时，能对由非正常因素引起的运行错误给出适当的处理或信息提示，使软件运行正常结束

从解释中可以看出，简单理解，所谓**的`容错处理`其实就是捕获异常了，不让异常影响系统的正常运行，正如`java`中的`try catch`一样。**

而在SpringCloud微服务调用中，自身异常可自行处理外，对于依赖的服务若发生错误，或者**调用异常**，或者调用时间过长等原因时，避免长时间等待，造成系统资源耗尽。  
**一般上都会通过`设置请求的超时时间`，如`http`请求中的`ConnectTimeout`和`ReadTimeout`；再或者就是使用`熔断器`模式，隔离问题服务，防止级联错误等。**

**为了防止服务之间的调用异常造成的连锁反应，在SpringCloud中提供了Hystrix组件来实现服务调用异常的处理，或对高并发情况下的服务降级处理 。**

# 什么是Hystrix #

>  在分布式系统中，服务与服务之间依赖错综复杂，一种不可避免的情况就是某些服务将会出现失败。
> 
> **Hystrix是一个库，它提供了服务与服务之间的容错功能**，主要体现在延迟容错和容错，从而做到控制分布式系统中的联动故障。**Hystrix通过隔离服务的访问点，阻止联动故障，并提供故障的解决方案，从而提高了这个分布式系统的弹性。**

## **Hystrix容错机制：** ##

>  *  **包裹请求**：**使用HystrixCommand**包裹对依赖的调用逻辑，每个命令在独立线程中执行，这是用到了设计模式“命令模式”。
>  *  **跳闸机制**：当某服务的错误率超过一定阈值时，**Hystrix可以自动或手动跳闸，停止请求该服务一段时间**。
>  *  **资源隔离**：**Hystrix为每个依赖都维护了一个小型的线程池**，如果该线程池已满，发往该依赖的请求就被立即拒绝，而不是排队等候，从而加速判定失败。
>  *  **监控**：Hystrix可以**近乎实时的监控运行指标和配置的变化**。如成功、失败、超时、被拒绝的请求等。
>  *  **回退机制**：当请求失败、超时、被拒绝，或当断路器打开时，执行回退逻辑。回退逻辑可自定义。
>  *  **自我修复**：断路器打开一段时间后，会自动进入半开状态，**断路器打开、关闭、半开的逻辑转换**。

下图就是`Hystrix`的回退策略，防止级联故障。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70][] Hystrix fallback prevents cascading failures

# Hystrix的简单使用 #

1.要使用 Hystrix熔断机制处理引入它本身的依赖之外，我们需要在**主程序配置类上引入 @EnableHystrix 标签 开启Hystrix功能**，如下

@EnableHystrix
    @EnableEurekaClient
    @SpringBootApplication
    ...
    public class ConsumerApplication {

2.开启Hystrix熔断机制后**，对方法进行熔断处理**

@Service
    public class HelloService {
    
        @Autowired
        private RestTemplate restTemplate;
    
        //该注解对该方法创建了熔断器的功能，并指定了fallbackMethod熔断方法
        @HystrixCommand(fallbackMethod = "hiError")
        public String hiService(String name){
            //调用接口进行消费
            String result = restTemplate.getForObject("http://PRODUCER/hello?name="+name,String.class);
            return result;
        }
        public String hiError(String name) {
            return "hi,"+name+"error!";
        }
    }

当hiService方法第调用异常，**会触发 fallbackMethod执行的hiError方法做成一些补救处理**

# Hystrix的工作原理 #

下图显示通过Hystrix向服务依赖关系发出请求时会发生什么：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 1][] Hystrix向服务依赖关系发出请求时会发生什么

具体将从以下几个方面进行描述：

## 1.构建一个HystrixCommand或者HystrixObservableCommand 对象。 ##

第一步是构建一个HystrixCommand或HystrixObservableCommand对象来表示你对依赖关系的请求。 其中构造函数需要和请求时的参数一致。

构造HystrixCommand对象，如果依赖关系预期返回单个响应。 可以这样写：

HystrixCommand command = new HystrixCommand(arg1, arg2);

同理，可以构建HystrixObservableCommand ：

HystrixObservableCommand command = new HystrixObservableCommand(arg1, arg2);

## 2.执行Command ##

通过使用Hystrix命令对象的以下四种方法之一，可以执行该命令有四种方法（前两种方法仅适用于简单的HystrixCommand对象，并不适用于HystrixObservableCommand）：

>  *  execute()–阻塞，，然后返回从依赖关系接收到的单个响应（或者在发生错误时抛出异常）
>  *  queue()–返回一个可以从依赖关系获得单个响应的future 对象
>  *  observe()–订阅Observable代表依赖关系的响应，并返回一个Observable，该Observable会复制该来源Observable
>  *  toObservable() –返回一个Observable，当您订阅它时，将执行Hystrix命令并发出其响应

K             value   = command.execute();
    Future<K>     fValue  = command.queue();
    Observable<K> ohValue = command.observe();         
    Observable<K> ocValue = command.toObservable();

同步调用execute（）调用queue（）.get（）. queue（）依次调用toObservable（）.toBlocking（）.toFuture（）。 这就是说，**最终每个HystrixCommand都由一个Observable实现支持**，甚至是那些旨在返回单个简单值的命令。

## 3.响应是否有缓存？ ##

如果为该命令启用请求缓存，并且如果缓存中对该请求的响应可用，则此缓存响应将立即以“可观察”的形式返回。

## 4.断路器是否打开？ ##

当您执行该命令时，Hystrix将检查断路器以查看电路是否打开。

如果电路打开（或“跳闸”），则Hystrix将不会执行该命令，但会将流程路由到（8）获取回退。

如果电路关闭，则流程进行到（5）以检查是否有可用于运行命令的容量。

## 5.线程池/队列/信号量是否已经满负载？ ##

如果与命令相关联的线程池和队列（或信号量，如果不在线程中运行）已满，则Hystrix将不会执行该命令，但将立即将流程路由到（8）获取回退。

## 6.HystrixObservableCommand.construct() 或者 HystrixCommand.run() ##

在这里，Hystrix通过您为此目的编写的方法调用对依赖关系的请求，其中之一是：

>  *  HystrixCommand.run（） – 返回单个响应或者引发异常
>  *  HystrixObservableCommand.construct（） – 返回一个发出响应的Observable或者发送一个onError通知

如果run（）或construct（）方法超出了命令的超时值，则该线程将抛出一个TimeoutException（或者如果命令本身没有在自己的线程中运行，则会产生单独的计时器线程）。 在这种情况下，Hystrix将响应通过8进行路由。获取Fallback，如果该方法不取消/中断，它会丢弃最终返回值run（）或construct（）方法。

请注意，没有办法强制潜在线程停止工作 – 最好的Hystrix可以在JVM上执行它来抛出一个InterruptedException。 如果由Hystrix包装的工作不处理InterruptedExceptions，Hystrix线程池中的线程将继续工作，尽管客户端已经收到了TimeoutException。 这种行为可能使Hystrix线程池饱和，尽管负载“正确地流失”。 大多数Java HTTP客户端库不会解释InterruptedExceptions。 因此，请确保在HTTP客户端上正确配置连接和读/写超时。

如果该命令没有引发任何异常并返回响应，则Hystrix在执行某些日志记录和度量报告后返回此响应。 在run（）的情况下，Hystrix返回一个Observable，发出单个响应，然后进行一个onCompleted通知; 在construct（）的情况下，Hystrix返回由construct（）返回的相同的Observable。

## 7.计算Circuit 的健康 ##

Hystrix向断路器报告成功，失败，拒绝和超时，该断路器维护了一系列的计算统计数据组。

它使用这些统计信息来确定电路何时“跳闸”，此时短路任何后续请求直到恢复时间过去，在首次检查某些健康检查之后，它再次关闭电路。

## 8.获取Fallback ##

当命令执行失败时，Hystrix试图恢复到你的回退：当construct（）或run（）（6.）抛出异常时，当命令由于电路断开而短路时（4.），当 命令的线程池和队列或信号量处于容量（5.），或者当命令超过其超时长度时。

编写Fallback ,它不一依赖于任何的网络依赖，从内存中获取获取通过其他的静态逻辑。如果你非要通过网络去获取Fallback,你可能需要些在获取服务的接口的逻辑上写一个HystrixCommand。

## 9.返回成功的响应 ##

如果Hystrix命令成功，它将以Observable的形式返回对呼叫者的响应或响应。 根据您在上述步骤2中调用命令的方式，此Observable可能会在返回给您之前进行转换：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 2][] 标题

>  *  execute（） – 以与.queue（）相同的方式获取Future，然后在此Future上调用get（）来获取Observable发出的单个值
>  *  queue（） – 将Observable转换为BlockingObservable，以便将其转换为Future，然后返回此未来
>  *  observe（） – 立即订阅Observable并启动执行命令的流程; 返回一个Observable，当您订阅它时，重播排放和通知
>  *  toObservable（） – 返回Observable不变; 您必须订阅它才能实际开始导致命令执行的流程

# Hystrix的源码解析 #

## 1. @EnableHystrix 开启Hystrix ##

沿着我上面的使用方式来跟踪一下 Hystrix的 源码。

首先我们看一下标签：**@EnableHystrix** ，他的作用从名字就能看出就是开启Hystrix ，我们看一下它的源码

@Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    @EnableCircuitBreaker
    public @interface EnableHystrix {
    
    }

它上面有一个注解：**@ EnableCircuitBreaker** (熔断器),**那么@ EnableHystrix标签的本质其实是@ EnableCircuitBreaker** ,我们看一下他的源码

/**
     * Annotation to enable a CircuitBreaker implementation.
     * http://martinfowler.com/bliki/CircuitBreaker.html
     * @author Spencer Gibb
     */
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    @Import(EnableCircuitBreakerImportSelector.class)
    public @interface EnableCircuitBreaker {
    
    }

@EnableCircuitBreaker标签引入了一个**@Import(EnableCircuitBreakerImportSelector.class) 类**，字面翻译的意思是开启熔断器的导入选择器 ，导入什么东西呢？看源码

/**
     * Import a single circuit breaker implementation Configuration
     * @author Spencer Gibb
     */
    @Order(Ordered.LOWEST_PRECEDENCE - 100)
    public class EnableCircuitBreakerImportSelector extends
            SpringFactoryImportSelector<EnableCircuitBreaker> {
    
        @Override
        protected boolean isEnabled() {
            return getEnvironment().getProperty(
                    "spring.cloud.circuit.breaker.enabled", Boolean.class, Boolean.TRUE);
        }
    
    }

**其实EnableCircuitBreakerImportSelector的作用就是去导入熔断器的配置** 。其实Spring中也有类似于JAVA SPI 的加载机制, 即会自动加载 jar包 spring-cloud-netflix-core 中的META-INF/spring.factories 中的Hystrix相关的自动配置类  
注：**SPI : 通过将服务的接口与实现分离以实现解耦，提高程序拓展性的机制，达到插拔式的效果 。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 3][]

HystrixCircuitBreakerConfiguration 就是针对于 Hystrix熔断器的配置

/**
     * @author Spencer Gibb
     * @author Christian Dupuis
     * @author Venil Noronha
     */
    @Configuration
    public class HystrixCircuitBreakerConfiguration {
    
        @Bean
        public HystrixCommandAspect hystrixCommandAspect() {
            return new HystrixCommandAspect();
        }
    
        @Bean
        public HystrixShutdownHook hystrixShutdownHook() {
            return new HystrixShutdownHook();
        }
    
        @Bean
        public HasFeatures hystrixFeature() {
            return HasFeatures.namedFeatures(new NamedFeature("Hystrix", HystrixCommandAspect.class));
        }
    ......

在该配置类中创建了 **HystrixCommandAspect**

/**
     * AspectJ aspect to process methods which annotated with {@link HystrixCommand} annotation.
     */
    @Aspect
    public class HystrixCommandAspect {
    
        private static final Map<HystrixPointcutType, MetaHolderFactory> META_HOLDER_FACTORY_MAP;
    
        static {
            META_HOLDER_FACTORY_MAP = ImmutableMap.<HystrixPointcutType, MetaHolderFactory>builder()
                    .put(HystrixPointcutType.COMMAND, new CommandMetaHolderFactory())
                    .put(HystrixPointcutType.COLLAPSER, new CollapserMetaHolderFactory())
                    .build();
        }
    
     //定义切点，切到 @HystrixCommand标签所在的方法  
      @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")
      public void hystrixCommandAnnotationPointcut() {
      }
    
      
    @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
      public void hystrixCollapserAnnotationPointcut() {
      }
    
    
       //针对切点：@hystrixCommand切点的处理
        @Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
        public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
        //获取到目标方法
        Method method = getMethodFromTarget(joinPoint);
            Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint);
            //判断方法上不能同时存在@HystrixCommand标签和HystrixCollapser标签
            if (method.isAnnotationPresent(HystrixCommand.class) && method.isAnnotationPresent(HystrixCollapser.class)) {
                throw new IllegalStateException("method cannot be annotated with HystrixCommand and HystrixCollapser " +
                        "annotations at the same time");
            }
            MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
            MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
           //把方法封装成 HystrixInvokable 
            HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
            ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
                    metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();
    
            Object result;
            try {
               // 通过CommandExecutor来执行方法
                if (!metaHolder.isObservable()) {
                    result = CommandExecutor.execute(invokable, executionType, metaHolder);
                } else {
                    result = executeObservable(invokable, executionType, metaHolder);
                }
            } catch (HystrixBadRequestException e) {
                throw e.getCause() != null ? e.getCause() : e;
            } catch (HystrixRuntimeException e) {
                throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
            }
            return result;

**HystrixCommandAspect 其实就是对 贴了@HystrixCommand标签的方法使用 Aop机制实现处理** 。代码中通过把目标方法封装成 HystrixInvokable对象，通过CommandExecutor工具来执行目标方法。

## ##

**HystrixInvokable是用来干嘛的？**看源码知道，**其实他是一个空方法的接口，他的目的只是用来标记可被执行**，那么他是如何创建的我们看代码HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);的create方法

public HystrixInvokable create(MetaHolder metaHolder) {
            HystrixInvokable executable;
            ...省略代码...
                executable = new GenericCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
            }
            return executable;
        }

其实是new了一个 GenericCommand 对象，很明显他们是实现关系，我们看一下关系图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 4][]

跟踪 GenericCommand 的源码

@ThreadSafe
    public class GenericCommand extends AbstractHystrixCommand<Object> {
        private static final Logger LOGGER = LoggerFactory.getLogger(GenericCommand.class);
    
        public GenericCommand(HystrixCommandBuilder builder) {
            super(builder);
        }
    
        protected Object run() throws Exception {
            LOGGER.debug("execute command: {}", this.getCommandKey().name());
            return this.process(new AbstractHystrixCommand<Object>.Action() {
                Object execute() {
                    return GenericCommand.this.getCommandAction().execute(GenericCommand.this.getExecutionType());
                }
            });
        }
    
        protected Object getFallback() {
            final CommandAction commandAction = this.getFallbackAction();
            if (commandAction != null) {
                try {
                    return this.process(new AbstractHystrixCommand<Object>.Action() {
                        Object execute() {
                            MetaHolder metaHolder = commandAction.getMetaHolder();
                            Object[] args = CommonUtils.createArgsForFallback(metaHolder, GenericCommand.this.getExecutionException());
                            return commandAction.executeWithArgs(metaHolder.getFallbackExecutionType(), args);
                        }
                    });
                } catch (Throwable var3) {
                    LOGGER.error(FallbackErrorMessageBuilder.create().append(commandAction, var3).build());
                    throw new FallbackInvocationException(ExceptionUtils.unwrapCause(var3));
                }
            } else {
                return super.getFallback();
            }
        }
    }

**它本身对目标方法的正常执行和对 fallback方法的 执行做了实现 。**  
GenericCommand.this.getCommandAction().execute(...)获取到目标方法并执行，底层会交给 MethodExecutionAction 使用反射去执行方法。

回到 HystrixCommandAspect的**methodsAnnotatedWithHystrixCommand**方法中，我们看下 **CommandExecutor.execute是如何执行的**

public class CommandExecutor {
        public CommandExecutor() {
        }
    
        public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
            Validate.notNull(invokable);
            Validate.notNull(metaHolder);
            switch(executionType) {
            //异步
            case SYNCHRONOUS:
                return castToExecutable(invokable, executionType).execute();
            //同步
            case ASYNCHRONOUS:
                HystrixExecutable executable = castToExecutable(invokable, executionType);
                if (metaHolder.hasFallbackMethodCommand() && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
                    return new FutureDecorator(executable.queue());
                }
    
                return executable.queue();
            case OBSERVABLE:
                HystrixObservable observable = castToObservable(invokable);
                return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
            default:
                throw new RuntimeException("unsupported execution type: " + executionType);
            }
        }
    
        private static HystrixExecutable castToExecutable(HystrixInvokable invokable, ExecutionType executionType) {
            if (invokable instanceof HystrixExecutable) {
                return (HystrixExecutable)invokable;
            } else {
                throw new RuntimeException("Command should implement " + HystrixExecutable.class.getCanonicalName() + " interface to execute in: " + executionType + " mode");
            }
        }

**这里有两种执行方式 SYNCHRONOUS 异步 ，ASYNCHRONOUS同步 ，**我们先看异步： castToExecutable(invokable, executionType).execute();

这里**代码把HystrixInvokable对象转成 HystrixExecutable并调用execute方法执行** ，跟踪execute方法进入HystrixCommand.execute方法中

public R execute() {
            try {
                return queue().get();
            } catch (Exception e) {
                throw Exceptions.sneakyThrow(decomposeException(e));
            }
        }
    --------------
     public Future<R> queue() {
            /*
             * The Future returned by Observable.toBlocking().toFuture() does not implement the
             * interruption of the execution thread when the "mayInterrupt" flag of Future.cancel(boolean) is set to true;
             * thus, to comply with the contract of Future, we must wrap around it.
             */
            final Future<R> delegate = toObservable().toBlocking().toFuture();
            
            final Future<R> f = new Future<R>() {
    
                @Override
                public boolean cancel(boolean mayInterruptIfRunning) {
                    if (delegate.isCancelled()) {
                        return false;
                    }
    
                    if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) {
                        /*
                         * The only valid transition here is false -> true. If there are two futures, say f1 and f2, created by this command
                         * (which is super-weird, but has never been prohibited), and calls to f1.cancel(true) and to f2.cancel(false) are
                         * issued by different threads, it's unclear about what value would be used by the time mayInterruptOnCancel is checked.
                         * The most consistent way to deal with this scenario is to say that if *any* cancellation is invoked with interruption,
                         * than that interruption request cannot be taken back.
                         */
                        interruptOnFutureCancel.compareAndSet(false, mayInterruptIfRunning);
                    }
    
                    final boolean res = delegate.cancel(interruptOnFutureCancel.get());
    
                    if (!isExecutionComplete() && interruptOnFutureCancel.get()) {
                        final Thread t = executionThread.get();
                        if (t != null && !t.equals(Thread.currentThread())) {
                            t.interrupt();
                        }
                    }
    
                    return res;
                }
    ....省略...

**在 HystrixCommand.execute方法中 其实是Future 来异步执行**，调用过程中会触发 GenericCommand来完成调用，执行完成后调用 Future.get()方法拿到执行结果 。HystrixCommand的queue方法实际上是调用了toObservable().toBlocking().toFuture()。

## 2.  **@HystrixCommand** ##

### 相关配置：HystrixCommandProperties ###

> 默认配置都在HystrixCommandProperties类中。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 5][]

**先看两个metrics收集的配置。**

*  **metrics.rollingStats.timeInMilliseconds**

> 表示滑动窗口的时间(the duration of the statistical rolling window)，默认10000(10s)，也是熔断器计算的基本单位。

*  **metrics.rollingStats.numBuckets**

> 滑动窗口的Bucket数量(the number of buckets the rolling statistical window is divided into)，默认10. 通过timeInMilliseconds和numBuckets可以计算出每个Bucket的时长。
> 
> metrics.rollingStats.timeInMilliseconds % metrics.rollingStats.numBuckets 必须等于 0，否则将抛异常。

**再看看熔断器的配置。**

*  **circuitBreaker.requestVolumeThreshold**

> 滑动窗口触发熔断的最小请求数。如果值是20，但滑动窗口的时间内请求数只有19，那即使19个请求全部失败，也不会熔断，必须达到这个值才行，否则样本太少，没有意义。

*  **circuitBreaker.sleepWindowInMilliseconds**

> 这个和熔断器自动恢复有关，为了检测后端服务是否恢复，可以放一个请求过去试探一下。sleepWindow指的发生熔断后，必须隔sleepWindow这么长的时间，才能放请求过去试探下服务是否恢复。默认是5s

*  **circuitBreaker.errorThresholdPercentage**

> 错误率阈值，表示达到熔断的条件。比如默认的50%，当一个滑动窗口内，失败率达到50%时就会触发熔断。

## HystrixCommand类 ##

> Hystrix Command 执行过程中，各种情况都以事件形式发出，再封装成特定的数据结构，最后汇入到事件流中(HystrixEventStream)。
> 
> 事件流提供了 observe() 方法，摇身一变，事件流把自己变成了一个数据源(各小溪汇入成河，消费者从河里取水)，其他消费者可以从这里获取数据，而 circuit-breaker 就是消费者之一。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 6][]

> 上面是官方完整的流程图，策略是：
> 
>  *  不断收集数据，达到条件就熔断；
>  *  熔断后拒绝所有请求一段时间(sleepWindow)；
>  *  然后放一个请求过去，如果请求成功，则关闭熔断器，否则继续打开熔断器。

查看HystrixCommand的父类AbstractCommand，其构造方法中定义了许多实例化hystrix所需的对象。

`AbstractCommand`这个类里，有个`initCircuitBreaker`这个方法。

private static HystrixCircuitBreaker initCircuitBreaker(boolean enabled, HystrixCircuitBreaker fromConstructor, HystrixCommandKey commandKey...) {
        // 如果启用了熔断器
        if (enabled) {
            // 若commandKey没有对应的CircuitBreaker,则创建
            if (fromConstructor == null) {
                return HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics);
            } else {
                // 如果有则返回现有的
                return fromConstructor;
            }
        } else {
            return new NoOpCircuitBreaker();
        }
    }

进入到`com.netflix.hystrix.HystrixCircuitBreaker.Factory.getInstance`观察

public static class Factory {
     // 维护一个ConcurrentHashMap保存熔断器
            private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap();
    
            public Factory() {
            }
    
            public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
                HystrixCircuitBreaker previouslyCached = (HystrixCircuitBreaker)circuitBreakersByCommand.get(key.name());
                // 判断ConcurrentHashMap中是否存有对应key的熔断器对象
                if (previouslyCached != null) {
                    return previouslyCached;
                } else {
                // 实例化并添加到ConcurrentHashMap里
                    HystrixCircuitBreaker cbForCommand = (HystrixCircuitBreaker)circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreaker.HystrixCircuitBreakerImpl(key, group, properties, metrics));
                    return cbForCommand == null ? (HystrixCircuitBreaker)circuitBreakersByCommand.get(key.name()) : cbForCommand;
                }
            }
    
            public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) {
                return (HystrixCircuitBreaker)circuitBreakersByCommand.get(key.name());
            }
    
            static void reset() {
                circuitBreakersByCommand.clear();
            }
        }
    }

接下来观察HystrixCircuitBreaker的实现类HystrixCircuitBreakerImpl

public static class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {
            // 熔断器的属性常量类
            private final HystrixCommandProperties properties;
           // 熔断器的熔断指标类
            private final HystrixCommandMetrics metrics;
            // 定义一个原子类型的布尔值状态
            private AtomicBoolean circuitOpen = new AtomicBoolean(false);
            private AtomicLong circuitOpenedOrLastTestedTime = new AtomicLong();
    
            protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
                this.properties = properties;
                this.metrics = metrics;
            }
    
    // 重置统计数据
            public void markSuccess() {
                if (this.circuitOpen.get() && this.circuitOpen.compareAndSet(true, false)) {
                    this.metrics.resetStream();
                }
    
            }
    
    // 是否放行请求
            public boolean allowRequest() {
            // 熔断器开启则不放行
                if ((Boolean)this.properties.circuitBreakerForceOpen().get()) {
                    return false;
                } else if ((Boolean)this.properties.circuitBreakerForceClosed().get()) {
                    this.isOpen();
                    return true;
                } else {
                    return !this.isOpen() || this.allowSingleTest();
                }
            }
    
    // 部分放行
            public boolean allowSingleTest() {
                long timeCircuitOpenedOrWasLastTested = this.circuitOpenedOrLastTestedTime.get();
                return this.circuitOpen.get() && System.currentTimeMillis() > timeCircuitOpenedOrWasLastTested + (long)(Integer)this.properties.circuitBreakerSleepWindowInMilliseconds().get() && this.circuitOpenedOrLastTestedTime.compareAndSet(timeCircuitOpenedOrWasLastTested, System.currentTimeMillis());
            }
    
    // 判断熔断器是否开启
            public boolean isOpen() {
                if (this.circuitOpen.get()) {
                    return true;
                } else {
                   // 根据采集到的数据进行判断是否开启
                    HealthCounts health = this.metrics.getHealthCounts();
                    if (health.getTotalRequests() < (long)(Integer)this.properties.circuitBreakerRequestVolumeThreshold().get()) {
                        return false;
                    } else if (health.getErrorPercentage() < (Integer)this.properties.circuitBreakerErrorThresholdPercentage().get()) {
                        return false;
                    } else if (this.circuitOpen.compareAndSet(false, true)) {
                        this.circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());
                        return true;
                    } else {
                        return true;
                    }
                }
            }
        }

### 如何订阅HystrixEventStream ###

本文最前面说到，HystrixEventStream提供了结构化的数据，提供了一个Observable对象，Hystrix只需要订阅它即可。

这HystrixCircuitBreaker接口实现类的构造器：

protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
        this.properties = properties;
        // 这是Command中的metrics对象,metrics对象也是commandKey维度的
        this.metrics = metrics;
        // !!!重点:订阅事件流
        Subscription s = subscribeToStream();
        activeSubscription.set(s);
    }
    // 订阅事件流, 前面打的比方: 小溪汇成的河, 各事件以结构化数据汇入了Stream中
    private Subscription subscribeToStream() {
        // HealthCountsStream是重点，下面会分析
        return metrics.getHealthCountsStream()
                .observe()
                // 利用数据统计的结果HealthCounts, 实现熔断器
                .subscribe(new Subscriber<HealthCounts>() {
                    @Override
                    public void onCompleted() {}
                    @Override
                    public void onError(Throwable e) {}
                    @Override
                    public void onNext(HealthCounts hc) {
                        // 检查是否达到最小请求数,默认20个; 未达到的话即使请求全部失败也不会熔断
                        if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
                            // 啥也不做
                        } else {
                            // 错误百分比未达到设定的阀值
                            if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
                            } else {
                                // 错误率过高, 进行熔断
                                if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
                                    circuitOpened.set(System.currentTimeMillis());
                                }
                            }
                        }
                    }
                });
    }

**HealthCounts** 属性如下，表示一个滑动窗口内的统计数据。

public static class HealthCounts {
        // rolling window 中请求总数量
        private final long totalCount;
        // 错误请求数(failure + success + timeout + threadPoolRejected + semaphoreRejected)
        private final long errorCount;
        // 错误率
        private final int errorPercentage;
    }

熔断器就是利用最终的统计结果HealthCounts来判断是否进行熔断。

### 熔断器状态变化 ###

熔断器有三种状态，如下：

enum Status {
        CLOSED, OPEN, HALF_OPEN;
    }

在Command的执行过程中，会调用HystrixCircuitBreaker的方法来更新状态。下面是几个重要的方法：

**命令执行时，判断熔断器是否打开**

// 是否允许执行
    public boolean attemptExecution() {
        // 熔断器配置了强制打开, 不允许执行命令
        if (properties.circuitBreakerForceOpen().get()) {
            return false;
        }
        // 熔断器配置了强制关闭, 允许执行
        if (properties.circuitBreakerForceClosed().get()) {
            return true;
        }
        // AtomicLong circuitOpened, -1是表示熔断器未打开
        if (circuitOpened.get() == -1) {
            return true;
        } else {
            // 熔断后,会拒绝所有命令一段时间(默认5s), 称为sleepWindow
            if (isAfterSleepWindow()) {
                // 过了sleepWindow后,将熔断器设置为"HALF_OPEN",允许第一个请求过去
                if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
                    return true;
                } else {
                    return false;
                }
            } else {
                return false;
            }
        }
    }

**当Command成功执行结束时，会调用HystrixCircuitBreaker.markSuccess()来标记执行成功.**

public void markSuccess() {
        // 如果是HALF_OPEN状态,则关闭熔断器
        if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
            // 重新开始统计metrics,抛弃所有原先的metrics信息
            metrics.resetStream();
            Subscription previousSubscription = activeSubscription.get();
            if (previousSubscription != null) {
                previousSubscription.unsubscribe();
            }
            Subscription newSubscription = subscribeToStream();
            activeSubscription.set(newSubscription);
            // circuitOpened设置为-1表示关闭熔断器
            circuitOpened.set(-1L);
        }
    }

**当Command执行失败时, 如果熔断器属于HALF\_OPEN状态,也就是熔断器刚过sleepWindow时间，尝试放一个请求过去，结果又失败了，于是马上打开熔断器，继续拒绝sleepWindow的时间。**

public void markNonSuccess() {
        if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {
            circuitOpened.set(System.currentTimeMillis());
        }
    }

下面是调用markNonSuccess()的地方，handleFallback是所有失败情况的处理者.

final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
        @Override
        public Observable<R> call(Throwable t) {
            circuitBreaker.markNonSuccess();
            Exception e = getExceptionFromThrowable(t);
            executionResult = executionResult.setExecutionException(e);
            // 线程池拒绝
            if (e instanceof RejectedExecutionException) {
                return handleThreadPoolRejectionViaFallback(e);
            // 超时
            } else if (t instanceof HystrixTimeoutException) {
                return handleTimeoutViaFallback();
            // Bad Request    
            } else if (t instanceof HystrixBadRequestException) {
                return handleBadRequestByEmittingError(e);
            } else {
                if (e instanceof HystrixBadRequestException) {
                    eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
                    return Observable.error(e);
                }
                return handleFailureViaFallback(e);
            }
        }
    };

**Circuit-Breaker的设计、实现都很有意思：**

>  *  滴水成河，收集每个命令的执行情况，汇总后通过滑动窗口，不断动态计算最新统计数据，基于统计数据来开启熔断器
>  *  巧妙的利用RxJava的window()函数来汇总数据，先汇总为Bucket， N Bucket组成Rolling Window
>  *  使用sleepWindow + 尝试机制，自动恢复 “供电”

参考链接：

https://www.jianshu.com/p/07d45032bdff

[https://www.javajike.com/article/2060.html][https_www.javajike.com_article_2060.html]

[https://www.kancloud.cn/fymod/springcloud2/784132][https_www.kancloud.cn_fymod_springcloud2_784132]

[https://www.cnblogs.com/handsome1013/p/11193746.html][https_www.cnblogs.com_handsome1013_p_11193746.html]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200130200951293.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200130204300296.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200130205440428.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/2020013021045662.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200130212442333.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200130222904658.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200130224342196.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZseTkxMDkwNQ==,size_16,color_FFFFFF,t_70
[https_www.javajike.com_article_2060.html]: https://www.javajike.com/article/2060.html
[https_www.kancloud.cn_fymod_springcloud2_784132]: https://www.kancloud.cn/fymod/springcloud2/784132
[https_www.cnblogs.com_handsome1013_p_11193746.html]: https://www.cnblogs.com/handsome1013/p/11193746.html