Hystrix原理与实战

 

总括：

 

 

注解的作用

1，用于标记处理

2，结合切面使用，打包配置属性生成代理对象，用反射等实现配置效果

 

 

 

下面的代码示例相当于静态代理以某一个类实现熔断举例，在具体方法上，类上用注解属性开启是动态代理的方式,生成的代码和静态编码一样

每个熔断器代理类中：构造函数中设置阈值，metrics中计数，超过阈值就打开熔断器，回退

熔断器的打开：

异常，熔断器打开都会走failback，但是走failback不一定是熔断器打开，

超过代理中构造函数设置的阈值就会打开熔断器.最终走failback

 

一些异常，虽然不会打开熔断器但是也是直接failback

 

 

线程池隔离---默认隔离

   为每一个服务创建一个线程池

   到了服务内部发起请求的线程和执行依赖服务的线程不是同一个线程

 

    用于控制并发，限流，防止雪崩，支持异步，线程切换，支持超时

 

   优点：把错误控制在线程池内，通过线程池的异常处理，可以定位具体服务具体问题

   缺点：增加了开销

 

一个服务一个threadPools，在threadPools中把请求放在一个线程池，命令执行又放在一个线程池

   final static ConcurrentHashMap<String, HystrixThreadPool> threadPools = new ConcurrentHashMap<String, HystrixThreadPool>();

...

if (!threadPools.containsKey(key)) {

    threadPools.put(key, new HystrixThreadPoolDefault(threadPoolKey, propertiesBuilder));

}

 

 

 

public class Service1HystrixCommand extends HystrixCommand<Response> {

  private Service1 service;

  private Request request;

 

  public Service1HystrixCommand(Service1 service, Request request){-------------------------------封装请求的线程

    supper(

      Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ServiceGroup"))

          .andCommandKey(HystrixCommandKey.Factory.asKey("servcie1query"))

          .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("service1ThreadPool"))

          .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter()

            .withCoreSize(20))//服务线程池数量

          .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()

            .withCircuitBreakerErrorThresholdPercentage(60)//熔断器关闭到打开阈值

            .withCircuitBreakerSleepWindowInMilliseconds(3000)//熔断器打开到关闭的时间窗长度

      ))

      this.service = service;

      this.request = request;

    );

  }

 

  @Override

  protected Response run(){----------------------------------------------执行命令的线程

    return service1.call(request);

  }

 

  @Override

  protected Response getFallback(){

    return Response.dummy();

  }

}

 

 

信号量隔离---相当于公用一个线程池，一个线程执行请求和命令服务

   所有的服务的请求在一个信号量中

   发起请求的线程和执行服务的线程都是一个线程

  

  用于控制并发，限流，防止雪崩

 

 

执行命令方法：

 

execute()

 同步阻塞

 

queue()

 异步非阻塞 future方式返回

 

observe()

 1，调用observe()会返回一个Observable对象

 2，用这个Observable对象的subscribe()方法完成事件注册，从而获取结果---当前请求来了，我注册下就是获取这个结果

 

 

 

 

hystrix容错的三种方式：

  公用核心参数：构造函数中设置都在构造函数中设置即可

     1，线程池，或信号量大小

     2，请求有超过多少个就计算错误率 默认 10

     3，熔断终端请求几秒后半开  默认5s

     5，请求错误超过多少个就打开熔断器 默认 50%

      public QueryOrderIdCommand(OrderServiceProvider orderServiceProvider) {

        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("orderService"))

                .andCommandKey(HystrixCommandKey.Factory.asKey("queryByOrderId"))

                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()

                        .withCircuitBreakerRequestVolumeThreshold(10)//至少有10个请求，熔断器才进行错误率的计算

                        .withCircuitBreakerSleepWindowInMilliseconds(5000)//熔断器中断请求5秒后会进入半打开状态,放部分流量过去重试

                        .withCircuitBreakerErrorThresholdPercentage(50)//错误率达到50开启熔断保护

                        .withExecutionTimeoutEnabled(true))

                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties

                        .Setter().withCoreSize(10)));

        this.orderServiceProvider = orderServiceProvider;

    }

 

 

public QueryByOrderIdCommandSemaphore(OrderServiceProvider orderServiceProvider) {

super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("orderService"))

.andCommandKey(HystrixCommandKey.Factory.asKey("queryByOrderId"))

.andCommandPropertiesDefaults(HystrixCommandProperties.Setter()

.withCircuitBreakerRequestVolumeThreshold(10)////至少有10个请求，熔断器才进行错误率的计算

.withCircuitBreakerSleepWindowInMilliseconds(5000)//熔断器中断请求5秒后会进入半打开状态,放部分流量过去重试

.withCircuitBreakerErrorThresholdPercentage(50)//错误率达到50开启熔断保护

.withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)

.withExecutionIsolationSemaphoreMaxConcurrentRequests(10)));//最大并发请求量

this.orderServiceProvider = orderServiceProvider;

    }

 

 

1，线程隔离，信号量隔离

   线程隔离一个服务请求一个线程池，请求和执行命令又分别开启线程池

     可快速失败，局部失败

     超量，超时，异常就打开熔断，最终都进入回退

 

     

 

 

 

2，熔断

   设置超时，和不重试

   可快速失败

   超量，超时，异常就打开熔断，最终都进入回退

 

核心参数：

    1，是否开启熔断器

    2，熔断器强制打开

    3，熔断器强制关闭

 

 

 

 

 

3，降级--回退

    降级的方式 

       1，failfast  系统出现任何异常，熔断器打开，后直接抛出异常

       2，fail silent  系统出现异常，熔断器打开，调用指定的回退方法，返回null

       3，failback:static  系统异常，熔断器打开，返回默认的值--单个字符

       4，failback:stubbed  系统异常，熔断器打开，返回一个对象

       5，failback:cache via network  系统异常，熔断器打开，用缓存历史数据

       6，Primary + Secondary with Fallback。系统异常，熔断器打开切换主次线程---使用就的稳定版本

 

示例：

 

背景

分布式系统环境下，服务间类似依赖非常常见，一个业务调用通常依赖多个基础服务。如下图，对于同步调用，当库存服务不可用时，商品服务请求线程被阻塞，当有大批量请求调用库存服务时，最终可能导致整个商品服务资源耗尽，无法继续对外提供服务。并且这种不可用可能沿请求调用链向上传递，这种现象被称为雪崩效应。

雪崩效应常见场景

• 硬件故障：如服务器宕机，机房断电，光纤被挖断等。
• 流量激增：如异常流量，重试加大流量等。
• 缓存穿透：一般发生在应用重启，所有缓存失效时，以及短时间内大量缓存失效时。大量的缓存不命中，使请求直击后端服务，造成服务提供者超负荷运行，引起服务不可用。
• 程序BUG：如程序逻辑导致内存泄漏，JVM长时间FullGC等。
• 同步等待：服务间采用同步调用模式，同步等待造成的资源耗尽。

雪崩效应应对策略

针对造成雪崩效应的不同场景，可以使用不同的应对策略，没有一种通用所有场景的策略，参考如下：

• 硬件故障：多机房容灾、异地多活等。
• 流量激增：服务自动扩容、流量控制（限流、关闭重试）等。
• 缓存穿透：缓存预加载、缓存异步加载等。
• 程序BUG：修改程序bug、及时释放资源等。
• 同步等待：资源隔离、MQ解耦、不可用服务调用快速失败等。资源隔离通常指不同服务调用采用不同的线程池；不可用服务调用快速失败一般通过熔断器模式结合超时机制实现。

综上所述，如果一个应用不能对来自依赖的故障进行隔离，那该应用本身就处在被拖垮的风险中。 因此，为了构建稳定、可靠的分布式系统，我们的服务应当具有自我保护能力，当依赖服务不可用时，当前服务启动自我保护功能，从而避免发生雪崩效应。本文将重点介绍使用Hystrix解决同步等待的雪崩问题。

初探Hystrix

Hystrix [hɪst'rɪks]，中文含义是豪猪，因其背上长满棘刺，从而拥有了自我保护的能力。本文所说的Hystrix是Netflix开源的一款容错框架，同样具有自我保护能力。为了实现容错和自我保护，下面我们看看Hystrix如何设计和实现的。

Hystrix设计目标：

• 对来自依赖的延迟和故障进行防护和控制——这些依赖通常都是通过网络访问的
• 阻止故障的连锁反应
• 快速失败并迅速恢复
• 回退并优雅降级
• 提供近实时的监控与告警

Hystrix遵循的设计原则：

• 防止任何单独的依赖耗尽资源（线程）
• 过载立即切断并快速失败，防止排队
• 尽可能提供回退以保护用户免受故障
• 使用隔离技术（例如隔板，泳道和断路器模式）来限制任何一个依赖的影响
• 通过近实时的指标，监控和告警，确保故障被及时发现
• 通过动态修改配置属性，确保故障及时恢复
• 防止整个依赖客户端执行失败，而不仅仅是网络通信

Hystrix如何实现这些设计目标？

• 使用命令模式将所有对外部服务（或依赖关系）的调用包装在HystrixCommand或HystrixObservableCommand对象中，并将该对象放在单独的线程中执行；
• 每个依赖都维护着一个线程池（或信号量），线程池被耗尽则拒绝请求（而不是让请求排队）。
• 记录请求成功，失败，超时和线程拒绝。
• 服务错误百分比超过了阈值，熔断器开关自动打开，一段时间内停止对该服务的所有请求。
• 请求失败，被拒绝，超时或熔断时执行降级逻辑。
• 近实时地监控指标和配置的修改。

Hystrix入门

Hystrix简单示例

开始深入Hystrix原理之前，我们先简单看一个示例。

第一步，继承HystrixCommand实现自己的command，在command的构造方法中需要配置请求被执行需要的参数，并组合实际发送请求的对象，代码如下：

1.  
   public class QueryOrderIdCommand extends HystrixCommand<Integer> {
2.  
   private final static Logger logger = LoggerFactory.getLogger(QueryOrderIdCommand.class);
3.  
   private OrderServiceProvider orderServiceProvider;
4.  
    
5.  
   public QueryOrderIdCommand(OrderServiceProvider orderServiceProvider) {
6.  
   super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("orderService"))
7.  
   .andCommandKey(HystrixCommandKey.Factory.asKey("queryByOrderId"))
8.  
   .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
9.  
   .withCircuitBreakerRequestVolumeThreshold(10)//至少有10个请求，熔断器才进行错误率的计算
10.  
    .withCircuitBreakerSleepWindowInMilliseconds(5000)//熔断器中断请求5秒后会进入半打开状态,放部分流量过去重试
11.  
    .withCircuitBreakerErrorThresholdPercentage(50)//错误率达到50开启熔断保护
12.  
    .withExecutionTimeoutEnabled(true))
13.  
    .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties
14.  
    .Setter().withCoreSize(10)));
15.  
    this.orderServiceProvider = orderServiceProvider;
16.  
    }
17.  
     
18.  
    @Override
19.  
    protected Integer run() {
20.  
    return orderServiceProvider.queryByOrderId();
21.  
    }
22.  
     
23.  
    @Override
24.  
    protected Integer getFallback() {
25.  
    return -1;
26.  
    }
27.  
    }

第二步，调用HystrixCommand的执行方法发起实际请求。

1.  
   @Test
2.  
   public void testQueryByOrderIdCommand() {
3.  
   Integer r = new QueryOrderIdCommand(orderServiceProvider).execute();
4.  
   logger.info("result:{}", r);
5.  
   }

Hystrix处理流程

Hystrix流程图如下：

                                图片来源Hystrix官网https://github.com/Netflix/Hystrix/wiki

Hystrix整个工作流如下：

1. 构造一个 HystrixCommand或HystrixObservableCommand对象，用于封装请求，并在构造方法配置请求被执行需要的参数；
2. 执行命令，Hystrix提供了4种执行命令的方法，后面详述；
3. 判断是否使用缓存响应请求，若启用了缓存，且缓存可用，直接使用缓存响应请求。Hystrix支持请求缓存，但需要用户自定义启动；
4. 判断熔断器是否打开，如果打开，跳到第8步；
5. 判断线程池/队列/信号量是否已满，已满则跳到第8步；
6. 执行HystrixObservableCommand.construct()或HystrixCommand.run()，如果执行失败或者超时，跳到第8步；否则，跳到第9步；
7. 统计熔断器监控指标；
8. 走Fallback备用逻辑
9. 返回请求响应

从流程图上可知道，第5步线程池/队列/信号量已满时，还会执行第7步逻辑，更新熔断器统计信息，而第6步无论成功与否，都会更新熔断器统计信息。

执行命令的几种方法

Hystrix提供了4种执行命令的方法，execute()和queue() 适用于HystrixCommand对象，而observe()和toObservable()适用于HystrixObservableCommand对象。

execute()

以同步堵塞方式执行run()，只支持接收一个值对象。hystrix会从线程池中取一个线程来执行run()，并等待返回值。

queue()

以异步非阻塞方式执行run()，只支持接收一个值对象。调用queue()就直接返回一个Future对象。可通过 Future.get()拿到run()的返回结果，但Future.get()是阻塞执行的。若执行成功，Future.get()返回单个返回值。当执行失败时，如果没有重写fallback，Future.get()抛出异常。

observe()

事件注册前执行run()/construct()，支持接收多个值对象，取决于发射源。调用observe()会返回一个hot Observable，也就是说，调用observe()自动触发执行run()/construct()，无论是否存在订阅者。

如果继承的是HystrixCommand，hystrix会从线程池中取一个线程以非阻塞方式执行run()；如果继承的是HystrixObservableCommand，将以调用线程阻塞执行construct()。

observe()使用方法：

1. 调用observe()会返回一个Observable对象
2. 调用这个Observable对象的subscribe()方法完成事件注册，从而获取结果

toObservable()

事件注册后执行run()/construct()，支持接收多个值对象，取决于发射源。调用toObservable()会返回一个cold Observable，也就是说，调用toObservable()不会立即触发执行run()/construct()，必须有订阅者订阅Observable时才会执行。

如果继承的是HystrixCommand，hystrix会从线程池中取一个线程以非阻塞方式执行run()，调用线程不必等待run()；如果继承的是HystrixObservableCommand，将以调用线程堵塞执行construct()，调用线程需等待construct()执行完才能继续往下走。

toObservable()使用方法：

1. 调用observe()会返回一个Observable对象
2. 调用这个Observable对象的subscribe()方法完成事件注册，从而获取结果

需注意的是，HystrixCommand也支持toObservable()和observe()，但是即使将HystrixCommand转换成Observable，它也只能发射一个值对象。只有HystrixObservableCommand才支持发射多个值对象。

几种方法的关系

• execute()实际是调用了queue().get()
• queue()实际调用了toObservable().toBlocking().toFuture()
• observe()实际调用toObservable()获得一个cold Observable，再创建一个ReplaySubject对象订阅Observable，将源Observable转化为hot Observable。因此调用observe()会自动触发执行run()/construct()。

Hystrix总是以Observable的形式作为响应返回，不同执行命令的方法只是进行了相应的转换。

Hystrix容错

Hystrix的容错主要是通过添加容许延迟和容错方法，帮助控制这些分布式服务之间的交互。 还通过隔离服务之间的访问点，阻止它们之间的级联故障以及提供回退选项来实现这一点，从而提高系统的整体弹性。Hystrix主要提供了以下几种容错方法：

• 资源隔离
• 熔断
• 降级

下面我们详细谈谈这几种容错机制。

资源隔离

资源隔离主要指对线程的隔离。Hystrix提供了两种线程隔离方式：线程池和信号量。

线程隔离-线程池

Hystrix通过命令模式对发送请求的对象和执行请求的对象进行解耦，将不同类型的业务请求封装为对应的命令请求。如订单服务查询商品，查询商品请求->商品Command；商品服务查询库存，查询库存请求->库存Command。并且为每个类型的Command配置一个线程池，当第一次创建Command时，根据配置创建一个线程池，并放入ConcurrentHashMap，如商品Command：

1.  
   final static ConcurrentHashMap<String, HystrixThreadPool> threadPools = new ConcurrentHashMap<String, HystrixThreadPool>();
2.  
   ...
3.  
   if (!threadPools.containsKey(key)) {
4.  
   threadPools.put(key, new HystrixThreadPoolDefault(threadPoolKey, propertiesBuilder));
5.  
   }

后续查询商品的请求创建Command时，将会重用已创建的线程池。线程池隔离之后的服务依赖关系：

通过将发送请求线程与执行请求的线程分离，可有效防止发生级联故障。当线程池或请求队列饱和时，Hystrix将拒绝服务，使得请求线程可以快速失败，从而避免依赖问题扩散。

线程池隔离优缺点

优点：

• 保护应用程序以免受来自依赖故障的影响，指定依赖线程池饱和不会影响应用程序的其余部分。
• 当引入新客户端lib时，即使发生问题，也是在本lib中，并不会影响到其他内容。
• 当依赖从故障恢复正常时，应用程序会立即恢复正常的性能。
• 当应用程序一些配置参数错误时，线程池的运行状况会很快检测到这一点（通过增加错误，延迟，超时，拒绝等），同时可以通过动态属性进行实时纠正错误的参数配置。
• 如果服务的性能有变化，需要实时调整，比如增加或者减少超时时间，更改重试次数，可以通过线程池指标动态属性修改，而且不会影响到其他调用请求。
• 除了隔离优势外，hystrix拥有专门的线程池可提供内置的并发功能，使得可以在同步调用之上构建异步门面（外观模式），为异步编程提供了支持（Hystrix引入了Rxjava异步框架）。

注意：尽管线程池提供了线程隔离，我们的客户端底层代码也必须要有超时设置或响应线程中断，不能无限制的阻塞以致线程池一直饱和。

缺点：

线程池的主要缺点是增加了计算开销。每个命令的执行都在单独的线程完成，增加了排队、调度和上下文切换的开销。因此，要使用Hystrix，就必须接受它带来的开销，以换取它所提供的好处。

通常情况下，线程池引入的开销足够小，不会有重大的成本或性能影响。但对于一些访问延迟极低的服务，如只依赖内存缓存，线程池引入的开销就比较明显了，这时候使用线程池隔离技术就不适合了，我们需要考虑更轻量级的方式，如信号量隔离。

线程隔离-信号量

上面提到了线程池隔离的缺点，当依赖延迟极低的服务时，线程池隔离技术引入的开销超过了它所带来的好处。这时候可以使用信号量隔离技术来代替，通过设置信号量来限制对任何给定依赖的并发调用量。下图说明了线程池隔离和信号量隔离的主要区别：

                            图片来源Hystrix官网https://github.com/Netflix/Hystrix/wiki

使用线程池时，发送请求的线程和执行依赖服务的线程不是同一个，而使用信号量时，发送请求的线程和执行依赖服务的线程是同一个，都是发起请求的线程。先看一个使用信号量隔离线程的示例：

1.  
   public class QueryByOrderIdCommandSemaphore extends HystrixCommand<Integer> {
2.  
   private final static Logger logger = LoggerFactory.getLogger(QueryByOrderIdCommandSemaphore.class);
3.  
   private OrderServiceProvider orderServiceProvider;
4.  
    
5.  
   public QueryByOrderIdCommandSemaphore(OrderServiceProvider orderServiceProvider) {
6.  
   super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("orderService"))
7.  
   .andCommandKey(HystrixCommandKey.Factory.asKey("queryByOrderId"))
8.  
   .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
9.  
   .withCircuitBreakerRequestVolumeThreshold(10)////至少有10个请求，熔断器才进行错误率的计算
10.  
    .withCircuitBreakerSleepWindowInMilliseconds(5000)//熔断器中断请求5秒后会进入半打开状态,放部分流量过去重试
11.  
    .withCircuitBreakerErrorThresholdPercentage(50)//错误率达到50开启熔断保护
12.  
    .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)
13.  
    .withExecutionIsolationSemaphoreMaxConcurrentRequests(10)));//最大并发请求量
14.  
    this.orderServiceProvider = orderServiceProvider;
15.  
    }
16.  
     
17.  
    @Override
18.  
    protected Integer run() {
19.  
    return orderServiceProvider.queryByOrderId();
20.  
    }
21.  
     
22.  
    @Override
23.  
    protected Integer getFallback() {
24.  
    return -1;
25.  
    }
26.  
    }

由于Hystrix默认使用线程池做线程隔离，使用信号量隔离需要显示地将属性execution.isolation.strategy设置为ExecutionIsolationStrategy.SEMAPHORE，同时配置信号量个数，默认为10。客户端需向依赖服务发起请求时，首先要获取一个信号量才能真正发起调用，由于信号量的数量有限，当并发请求量超过信号量个数时，后续的请求都会直接拒绝，进入fallback流程。

信号量隔离主要是通过控制并发请求量，防止请求线程大面积阻塞，从而达到限流和防止雪崩的目的。

线程隔离总结

线程池和信号量都可以做线程隔离，但各有各的优缺点和支持的场景，对比如下：

	线程切换	支持异步	支持超时	支持熔断	限流	开销
信号量	否	否	否	是	是	小
线程池	是	是	是	是	是	大

线程池和信号量都支持熔断和限流。相比线程池，信号量不需要线程切换，因此避免了不必要的开销。但是信号量不支持异步，也不支持超时，也就是说当所请求的服务不可用时，信号量会控制超过限制的请求立即返回，但是已经持有信号量的线程只能等待服务响应或从超时中返回，即可能出现长时间等待。线程池模式下，当超过指定时间未响应的服务，Hystrix会通过响应中断的方式通知线程立即结束并返回。

熔断

熔断器简介

现实生活中，可能大家都有注意到家庭电路中通常会安装一个保险盒，当负载过载时，保险盒中的保险丝会自动熔断，以保护电路及家里的各种电器，这就是熔断器的一个常见例子。Hystrix中的熔断器(Circuit Breaker)也是起类似作用，Hystrix在运行过程中会向每个commandKey对应的熔断器报告成功、失败、超时和拒绝的状态，熔断器维护并统计这些数据，并根据这些统计信息来决策熔断开关是否打开。如果打开，熔断后续请求，快速返回。隔一段时间（默认是5s）之后熔断器尝试半开，放入一部分流量请求进来，相当于对依赖服务进行一次健康检查，如果请求成功，熔断器关闭。

熔断器配置

Circuit Breaker主要包括如下6个参数：

1、circuitBreaker.enabled

是否启用熔断器，默认是TRUE。
2 、circuitBreaker.forceOpen

熔断器强制打开，始终保持打开状态，不关注熔断开关的实际状态。默认值FLASE。
3、circuitBreaker.forceClosed
熔断器强制关闭，始终保持关闭状态，不关注熔断开关的实际状态。默认值FLASE。

4、circuitBreaker.errorThresholdPercentage
错误率，默认值50%，例如一段时间（10s）内有100个请求，其中有54个超时或者异常，那么这段时间内的错误率是54%，大于了默认值50%，这种情况下会触发熔断器打开。

5、circuitBreaker.requestVolumeThreshold

默认值20。含义是一段时间内至少有20个请求才进行errorThresholdPercentage计算。比如一段时间了有19个请求，且这些请求全部失败了，错误率是100%，但熔断器不会打开，总请求数不满足20。

6、circuitBreaker.sleepWindowInMilliseconds

半开状态试探睡眠时间，默认值5000ms。如：当熔断器开启5000ms之后，会尝试放过去一部分流量进行试探，确定依赖服务是否恢复。

熔断器工作原理

下图展示了HystrixCircuitBreaker的工作原理：

                                    图片来源Hystrix官网https://github.com/Netflix/Hystrix/wiki

熔断器工作的详细过程如下：

第一步，调用allowRequest()判断是否允许将请求提交到线程池

1. 如果熔断器强制打开，circuitBreaker.forceOpen为true，不允许放行，返回。
2. 如果熔断器强制关闭，circuitBreaker.forceClosed为true，允许放行。此外不必关注熔断器实际状态，也就是说熔断器仍然会维护统计数据和开关状态，只是不生效而已。

第二步，调用isOpen()判断熔断器开关是否打开

1. 如果熔断器开关打开，进入第三步，否则继续；
2. 如果一个周期内总的请求数小于circuitBreaker.requestVolumeThreshold的值，允许请求放行，否则继续；
3. 如果一个周期内错误率小于circuitBreaker.errorThresholdPercentage的值，允许请求放行。否则，打开熔断器开关，进入第三步。

第三步，调用allowSingleTest()判断是否允许单个请求通行，检查依赖服务是否恢复

1. 如果熔断器打开，且距离熔断器打开的时间或上一次试探请求放行的时间超过circuitBreaker.sleepWindowInMilliseconds的值时，熔断器器进入半开状态，允许放行一个试探请求；否则，不允许放行。

此外，为了提供决策依据，每个熔断器默认维护了10个bucket，每秒一个bucket，当新的bucket被创建时，最旧的bucket会被抛弃。其中每个blucket维护了请求成功、失败、超时、拒绝的计数器，Hystrix负责收集并统计这些计数器。

熔断器测试

1、以QueryOrderIdCommand为测试command

2、配置orderServiceProvider不重试且500ms超时

1.  
   <dubbo:reference id="orderServiceProvider" interface="com.huang.provider.OrderServiceProvider"
2.  
   timeout="500" retries="0"/>

3、OrderServiceProviderImpl实现很简单，前10个请求，服务端休眠600ms，使得客户端调用超时。

1.  
   @Service
2.  
   public class OrderServiceProviderImpl implements OrderServiceProvider {
3.  
   private final static Logger logger = LoggerFactory.getLogger(OrderServiceProviderImpl.class);
4.  
   private AtomicInteger OrderIdCounter = new AtomicInteger(0);
5.  
    
6.  
   @Override
7.  
   public Integer queryByOrderId() {
8.  
   int c = OrderIdCounter.getAndIncrement();
9.  
   if (logger.isDebugEnabled()) {
10.  
    logger.debug("OrderIdCounter:{}", c);
11.  
    }
12.  
    if (c < 10) {
13.  
    try {
14.  
    Thread.sleep(600);
15.  
    } catch (InterruptedException e) {
16.  
    }
17.  
    }
18.  
    return c;
19.  
    }
20.  
     
21.  
    @Override
22.  
    public void reset() {
23.  
    OrderIdCounter.getAndSet(0);
24.  
    }
25.  
    }

4、单测代码

1.  
   @Test
2.  
   public void testExecuteCommand() throws InterruptedException {
3.  
   orderServiceProvider.reset();
4.  
   int i = 1;
5.  
   for (; i < 15; i++) {
6.  
   HystrixCommand<Integer> command = new QueryByOrderIdCommand(orderServiceProvider);
7.  
   Integer r = command.execute();
8.  
   String method = r == -1 ? "fallback" : "run";
9.  
   logger.info("call {} times,result:{},method:{},isCircuitBreakerOpen:{}", i, r, method, command.isCircuitBreakerOpen());
10.  
    }
11.  
    //等待6s，使得熔断器进入半打开状态
12.  
    Thread.sleep(6000);
13.  
    for (; i < 20; i++) {
14.  
    HystrixCommand<Integer> command = new QueryByOrderIdCommand(orderServiceProvider);
15.  
    Integer r = command.execute();
16.  
    String method = r == -1 ? "fallback" : "run";
17.  
    logger.info("call {} times,result:{},method:{},isCircuitBreakerOpen:{}", i, r, method, command.isCircuitBreakerOpen());
18.  
    }
19.  
    }

5、输出结果

1.  
   2018-02-07 11:38:36,056 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 1 times,result:-1,method:fallback,isCircuitBreakerOpen:false
2.  
   2018-02-07 11:38:36,564 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 2 times,result:-1,method:fallback,isCircuitBreakerOpen:false
3.  
   2018-02-07 11:38:37,074 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 3 times,result:-1,method:fallback,isCircuitBreakerOpen:false
4.  
   2018-02-07 11:38:37,580 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 4 times,result:-1,method:fallback,isCircuitBreakerOpen:false
5.  
   2018-02-07 11:38:38,089 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 5 times,result:-1,method:fallback,isCircuitBreakerOpen:false
6.  
   2018-02-07 11:38:38,599 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 6 times,result:-1,method:fallback,isCircuitBreakerOpen:false
7.  
   2018-02-07 11:38:39,109 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 7 times,result:-1,method:fallback,isCircuitBreakerOpen:false
8.  
   2018-02-07 11:38:39,622 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 8 times,result:-1,method:fallback,isCircuitBreakerOpen:false
9.  
   2018-02-07 11:38:40,138 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 9 times,result:-1,method:fallback,isCircuitBreakerOpen:false
10.  
    2018-02-07 11:38:40,647 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 10 times,result:-1,method:fallback,isCircuitBreakerOpen:true
11.  
    2018-02-07 11:38:40,651 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 11 times,result:-1,method:fallback,isCircuitBreakerOpen:true
12.  
    2018-02-07 11:38:40,653 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 12 times,result:-1,method:fallback,isCircuitBreakerOpen:true
13.  
    2018-02-07 11:38:40,656 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 13 times,result:-1,method:fallback,isCircuitBreakerOpen:true
14.  
    2018-02-07 11:38:40,658 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:36 call 14 times,result:-1,method:fallback,isCircuitBreakerOpen:true
15.  
    2018-02-07 11:38:46,671 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:44 call 15 times,result:10,method:run,isCircuitBreakerOpen:false
16.  
    2018-02-07 11:38:46,675 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:44 call 16 times,result:11,method:run,isCircuitBreakerOpen:false
17.  
    2018-02-07 11:38:46,680 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:44 call 17 times,result:12,method:run,isCircuitBreakerOpen:false
18.  
    2018-02-07 11:38:46,685 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:44 call 18 times,result:13,method:run,isCircuitBreakerOpen:false
19.  
    2018-02-07 11:38:46,691 INFO [main] com.huang.test.command.QueryByOrderIdCommandTest:testExecuteCommand:44 call 19 times,result:14,method:run,isCircuitBreakerOpen:false

前9个请求调用超时，走fallback逻辑；

10-14个请求，熔断器开关打开，直接快速失败走fallback逻辑；

15-19个请求，熔断器进入半开状态，放行一个试探请求调用成功，熔断器关闭，后续请求恢复。

回退降级

降级，通常指务高峰期，为了保证核心服务正常运行，需要停掉一些不太重要的业务，或者某些服务不可用时，执行备用逻辑从故障服务中快速失败或快速返回，以保障主体业务不受影响。Hystrix提供的降级主要是为了容错，保证当前服务不受依赖服务故障的影响，从而提高服务的健壮性。要支持回退或降级处理，可以重写HystrixCommand的getFallBack方法或HystrixObservableCommand的resumeWithFallback方法。

Hystrix在以下几种情况下会走降级逻辑：

• 执行construct()或run()抛出异常
• 熔断器打开导致命令短路
• 命令的线程池和队列或信号量的容量超额，命令被拒绝
• 命令执行超时

降级回退方式

Fail Fast 快速失败

快速失败是最普通的命令执行方法，命令没有重写降级逻辑。 如果命令执行发生任何类型的故障，它将直接抛出异常。

Fail Silent 无声失败

指在降级方法中通过返回null，空Map，空List或其他类似的响应来完成。

1.  
   @Override
2.  
   protected Integer getFallback() {
3.  
   return null;
4.  
   }
5.  
    
6.  
   @Override
7.  
   protected List<Integer> getFallback() {
8.  
   return Collections.emptyList();
9.  
   }
10.  
     
11.  
    @Override
12.  
    protected Observable<Integer> resumeWithFallback() {
13.  
    return Observable.empty();
14.  
    }

Fallback: Static

指在降级方法中返回静态默认值。 这不会导致服务以“无声失败”的方式被删除，而是导致默认行为发生。如：应用根据命令执行返回true / false执行相应逻辑，但命令执行失败，则默认为true

1.  
   @Override
2.  
   protected Boolean getFallback() {
3.  
   return true;
4.  
   }
5.  
   @Override
6.  
   protected Observable<Boolean> resumeWithFallback() {
7.  
   return Observable.just( true );
8.  
   }

Fallback: Stubbed

当命令返回一个包含多个字段的复合对象时，适合以Stubbed 的方式回退。

1.  
   @Override
2.  
   protected MissionInfo getFallback() {
3.  
   return new MissionInfo("missionName","error");
4.  
   }

Fallback: Cache via Network

有时，如果调用依赖服务失败，可以从缓存服务（如redis）中查询旧数据版本。由于又会发起远程调用，所以建议重新封装一个Command，使用不同的ThreadPoolKey，与主线程池进行隔离。

1.  
   @Override
2.  
   protected Integer getFallback() {
3.  
   return new RedisServiceCommand(redisService).execute();
4.  
   }

Primary + Secondary with Fallback

有时系统具有两种行为- 主要和次要，或主要和故障转移。主要和次要逻辑涉及到不同的网络调用和业务逻辑，所以需要将主次逻辑封装在不同的Command中，使用线程池进行隔离。为了实现主从逻辑切换，可以将主次command封装在外观HystrixCommand的run方法中，并结合配置中心设置的开关切换主从逻辑。由于主次逻辑都是经过线程池隔离的HystrixCommand，因此外观HystrixCommand可以使用信号量隔离，而没有必要使用线程池隔离引入不必要的开销。原理图如下：

                          图片来源Hystrix官网https://github.com/Netflix/Hystrix/wiki

主次模型的使用场景还是很多的。如当系统升级新功能时，如果新版本的功能出现问题，通过开关控制降级调用旧版本的功能。示例代码如下：

1.  
   public class CommandFacadeWithPrimarySecondary extends HystrixCommand<String> {
2.  
    
3.  
   private final static DynamicBooleanProperty usePrimary = DynamicPropertyFactory.getInstance().getBooleanProperty("primarySecondary.usePrimary", true);
4.  
    
5.  
   private final int id;
6.  
    
7.  
   public CommandFacadeWithPrimarySecondary(int id) {
8.  
   super(Setter
9.  
   .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))
10.  
    .andCommandKey(HystrixCommandKey.Factory.asKey("PrimarySecondaryCommand"))
11.  
    .andCommandPropertiesDefaults(
12.  
    // 由于主次command已经使用线程池隔离，Facade Command使用信号量隔离即可
13.  
    HystrixCommandProperties.Setter()
14.  
    .withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)));
15.  
    this.id = id;
16.  
    }
17.  
     
18.  
    @Override
19.  
    protected String run() {
20.  
    if (usePrimary.get()) {
21.  
    return new PrimaryCommand(id).execute();
22.  
    } else {
23.  
    return new SecondaryCommand(id).execute();
24.  
    }
25.  
    }
26.  
     
27.  
    @Override
28.  
    protected String getFallback() {
29.  
    return "static-fallback-" + id;
30.  
    }
31.  
     
32.  
    @Override
33.  
    protected String getCacheKey() {
34.  
    return String.valueOf(id);
35.  
    }
36.  
     
37.  
    private static class PrimaryCommand extends HystrixCommand<String> {
38.  
     
39.  
    private final int id;
40.  
     
41.  
    private PrimaryCommand(int id) {
42.  
    super(Setter
43.  
    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))
44.  
    .andCommandKey(HystrixCommandKey.Factory.asKey("PrimaryCommand"))
45.  
    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("PrimaryCommand"))
46.  
    .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(600)));
47.  
    this.id = id;
48.  
    }
49.  
     
50.  
    @Override
51.  
    protected String run() {
52.  
    return "responseFromPrimary-" + id;
53.  
    }
54.  
     
55.  
    }
56.  
     
57.  
    private static class SecondaryCommand extends HystrixCommand<String> {
58.  
     
59.  
    private final int id;
60.  
     
61.  
    private SecondaryCommand(int id) {
62.  
    super(Setter
63.  
    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))
64.  
    .andCommandKey(HystrixCommandKey.Factory.asKey("SecondaryCommand"))
65.  
    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("SecondaryCommand"))
66.  
    .andCommandPropertiesDefaults( HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(100)));
67.  
    this.id = id;
68.  
    }
69.  
     
70.  
    @Override
71.  
    protected String run() {
72.  
    return "responseFromSecondary-" + id;
73.  
    }
74.  
     
75.  
    }
76.  
     
77.  
    public static class UnitTest {
78.  
     
79.  
    @Test
80.  
    public void testPrimary() {
81.  
    HystrixRequestContext context = HystrixRequestContext.initializeContext();
82.  
    try {
83.  
    ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", true);
84.  
    assertEquals("responseFromPrimary-20", new CommandFacadeWithPrimarySecondary(20).execute());
85.  
    } finally {
86.  
    context.shutdown();
87.  
    ConfigurationManager.getConfigInstance().clear();
88.  
    }
89.  
    }
90.  
     
91.  
    @Test
92.  
    public void testSecondary() {
93.  
    HystrixRequestContext context = HystrixRequestContext.initializeContext();
94.  
    try {
95.  
    ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", false);
96.  
    assertEquals("responseFromSecondary-20", new CommandFacadeWithPrimarySecondary(20).execute());
97.  
    } finally {
98.  
    context.shutdown();
99.  
    ConfigurationManager.getConfigInstance().clear();
100.  
     }
101.  
     }
102.  
     }
103.  
     }

通常情况下，建议重写getFallBack或resumeWithFallback提供自己的备用逻辑，但不建议在回退逻辑中执行任何可能失败的操作。

总结

本文介绍了Hystrix及其工作原理，还介绍了Hystrix线程池隔离、信号量隔离和熔断器的工作原理，以及如何使用Hystrix的资源隔离，熔断和降级等技术实现服务容错，从而提高系统的整体健壮性。

 

 

 

Hystrix Metrics的实现===健康计数器：

 

Hystrix的Metrics中保存了当前服务的健康状况, 包括服务调用总次数和服务调用失败次数等. 根据Metrics的计数, 熔断器从而

能计算出当前服务的调用失败率, 用来和设定的阈值比较从而决定熔断器的状态切换逻辑. 因此Metrics的实现非常重要.

 

 

每个开启熔断的方法就是一个熔断器，每个熔断器会有一个metrics计数器，记录健康状况，

按事件分桶计数，每秒创建一个新的桶，去除旧的桶，每个熔断器的metrics默认10个桶

 

 

 

1.4之前的滑动窗口实现

Hystrix在这些版本中的使用自己定义的滑动窗口数据结构来记录当前时间窗的各种事件(成功,失败,超时,线程池拒绝等)的计数. 

事件产生时, 数据结构根据当前时间确定使用旧桶还是创建新桶来计数, 并在桶中对计数器经行修改. 

这些修改是多线程并发执行的, 代码中有不少加锁操作,逻辑较为复杂.

 

 

在这里插入图片描述

 

 

1.5之后的滑动窗口实现

Hystrix在这些版本中开始使用RxJava的Observable.window()实现滑动窗口.

RxJava的window使用后台线程创建新桶, 避免了并发创建桶的问题.

同时RxJava的单线程无锁特性也保证了计数变更时的线程安全. 从而使代码更加简洁. 

以下为我使用RxJava的window方法实现的一个简易滑动窗口Metrics, 短短几行代码便能完成统计功能,足以证明

RxJava的强大:

 

public void timeWindowTest() throws Exception{

  Observable<Integer> source = Observable.interval(50, TimeUnit.MILLISECONDS).map(i -> RandomUtils.nextInt(2));

  source.window(1, TimeUnit.SECONDS).subscribe(window -> {

    int[] metrics = new int[2];

    window.subscribe(i -> metrics[i]++,

      InternalObservableUtils.ERROR_NOT_IMPLEMENTED,

      () -> System.out.println("窗口Metrics:" + JSON.toJSONString(metrics)));

  });

  TimeUnit.SECONDS.sleep(3);

}

 

 

 

 

 

 

 

 

 

 

参考：

https://blog.csdn.net/loushuiyifan/article/details/82702522

 

 https://blog.csdn.net/qq_25484147/article/details/83375225