Hystrix断路器原理深度解析

文章目录

• • • • 1. HystrixCircuitBreaker
      • 2. 断路器Factory
      • 3. 实现类`HystrixCircuitBreakerImpl `
      • 4. 自动熔断
      • 5. 是否通过断路器
      • 5. HALF OPEN
      • 6. 滑动窗口原理

1. HystrixCircuitBreaker

Hystrix的熔断器实现在HystrixCircuitBreake中，此接口比较直观明了。整个HystrixCircuitBreaker接口一共有五个方法和三个静态类：

接口中的类:

• Factory: 维护了一个Hystrix命令和HystrixCircuitBreaker的关系的集合ConcurrentHashMap circuitBreakersByCommand。其中key通过HystrixCommandKey来定义，每一个Hystrix 命令都需要有一个Key来标识，同时根据这个Key可以找到对应的断路器实例。
• NoOpCircuitBreaker: 一个啥都不做的断路器，它允许所有请求通过，并且断路器始终处于闭合状态
• HystrixCircuitBreakerImpl:断路器的一个实现类。

接口中的方法

• allowRequest()方法表示是否允许指令执行，该方法在新版本中放弃
• isOpen()方法表示断路器是否为开启状态
• markSuccess()处于半开状态时，如果尝试请求成功，就调用这个方法。用于将断路器关闭
• markNonSuccess()处于半开状态时，如果尝试请求成功。 用来打开断路器
• attemptExecution()判断每个Hystrix命令的请求都通过它是否被执行。这是非幂等的，它可能会修改内部。旧版本中是allowRequest()。

2. 断路器Factory

Factory静态类相当于Circuit Breaker Factory，用于获取相应的HystrixCircuitBreaker。我们来看一下其实现：

class Factory {private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {// this should find it for all but the first timeHystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());if (previouslyCached != null) {return previouslyCached;}HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics));if (cbForCommand == null) {return circuitBreakersByCommand.get(key.name());} else {return cbForCommand;}}public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) {return circuitBreakersByCommand.get(key.name());}static void reset() {circuitBreakersByCommand.clear();}
}

简单的说就是Hystrix为每个commandKey都维护了一个熔断器，保持着对应的熔断器，所以当new XXXHystrixCommand()的时候依然能够保持着原来熔断器的状态。

正如前面分析，Hystrix在Factory类中维护了一个ConcurrentHashMap用于存储与每一个HystrixCommandKey相对应的HystrixCircuitBreaker。每当我们通过getInstance方法从中获取HystrixCircuitBreaker的时候，Hystrix首先会检查ConcurrentHashMap中有没有对应的缓存的断路器，如果有的话直接返回。如果没有的话就会新创建一个HystrixCircuitBreaker实例，将其添加到缓存中并且返回。关于断路器的实例化细节后面分析。

至于断路器的初始化时机，在Hystrix执行原理分析中提到过，在Hystrix实例化的时候，AbstractHystrix的构造函数中会调用断路器的Factory的上面的getInstance方法，一切都是非常合理自然~，下面看断路器的实现类。

3. 实现类HystrixCircuitBreakerImpl

先看实现类的代码：

class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {private final HystrixCommandProperties properties;private final HystrixCommandMetrics metrics;enum Status {CLOSED, OPEN, HALF_OPEN;}private final AtomicReference<Status> status = new AtomicReference<Status>(Status.CLOSED);private final AtomicLong circuitOpened = new AtomicLong(-1);private final AtomicReference<Subscription> activeSubscription = new AtomicReference<Subscription>(null);protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {this.properties = properties;this.metrics = metrics;//On a timer, this will set the circuit between OPEN/CLOSED as command executions occurSubscription s = subscribeToStream();activeSubscription.set(s);}
}

在该实现类中定义了断路器的五个核心属性:

• HystrixCommandProperties properties:断路器对应实例的属性集合对象
• HystrixCommandMetrics metrics:断路器最核心的东西，通过时间窗的形式，记录一段时间范围内（默认是10秒）的接口请求的健康状况（Command的执行状态，包括成功、失败、超时、线程池拒绝等）并得到对应的度量指标。
• AtomicReference status: 用来记录断路器的状态，默认是关闭状态
  • 断路器在状态变化时，使用了AtomicReference#compareAndSet来确保当条件满足时，只有一个请求能成功改变状态。
• AtomicLong circuitOpened:断路器打开的时间戳，用于判断休眠期的结束时间。默认-1，表示断路器未打开。
• AtomicReference activeSubscription: 这个是通过Rxjava实现的对HystrixCommandMetrics结果的观察者对象，当HystrixCommandMetrics值发生变化时会通知观察者。

HystrixCommandProperties properties该属性保存了断路器对应Command的所有属性。和线程池类似，HystrixCommandProperties也是保存到HashMap中，key为Command的key。下面主要介绍远断路器相关的属性。

1. circuitBreakerEnabled：熔断器是否启用，默认是true
2. circuitBreakerErrorThresholdPercentage 错误率阈值，默认50%，比如（10s）内有100个请求，其中有60个发生异常，那么这段时间的错误率是60，已经超过了错误率阈值，熔断器会自动打开。
3. circuitBreakerForceOpen：熔断器强制打开(可以强制设定状态)，始终保持打开状态，默认是false
4. circuitBreakerForceClosed：熔断器强制关闭，始终保持关闭状态，默认是false
5. circuitBreakerRequestVolumeThreshold：滑动窗口内（10s）的请求数阈值，只有达到了这个阈值，才有可能熔断。默认是20，如果这个时间段只有19个请求，就算全部失败了，也不会自动熔断。
6. circuitBreakerSleepWindowInMilliseconds：熔断器打开之后，为了能够自动恢复，每隔默认5000ms放一个请求过去，试探所依赖的服务是否恢复。

断路器HystrixCircuitBreaker有三个状态:

• CLOSED关闭状态：允许流量通过。
• OPEN打开状态：不允许流量通过，即处于降级状态，走降级逻辑。
• HALF_OPEN半开状态：允许某些流量通过，并关注这些流量的结果，如果出现超时、异常等情况，将进入OPEN状态，如果成功，那么将进入CLOSED状态。

4. 自动熔断

HystrixBreaker实例化的时候会订阅HystrixCommandMetrics的 HealthCountsStream，每当HealthCountsStream搜集到数据，都会触发 onNext方法，自动熔断就是这里发生。这个过程也可以叫做健康指标的订阅。

private Subscription subscribeToStream() {/** This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream*/return metrics.getHealthCountsStream().observe().subscribe(new Subscriber<HealthCounts>() {@Overridepublic void onNext(HealthCounts hc) {//首先校验的时在时间窗范围内的请求次数，如果低于阈值（默认是20），不做处理，如果高于阈值，则去判断接口请求的错误率if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {} else {//判断接口请求的错误率（阈值默认是50），如果高于这个值，则断路器打开if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {} else {//打开断路器，同时记录断路器开启时间if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {circuitOpened.set(System.currentTimeMillis());}}}}});
}

在onNext方法中，参数hc保存了当前接口在前10s之内(时间窗口)的请求状态（请求总数、失败数和失败率）,其主要逻辑是判断请求总数是否达到阈值requestVolumeThreshold，失败率是否达到阈值errorThresholdPercentage，如果都满足，说明接口的已经足够的不稳定，需要进行熔断，则设置status为熔断开启状态，并更新circuitOpened为当前时间戳，记录上次熔断开启的时间。

5. 是否通过断路器

每一个Command在execute执行后都会判断断路器是否可以通过，此逻辑实现在AbstractCommand#applyHystrixSemantics中：

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {//...省略部分代码if (circuitBreaker.attemptExecution()) {//...省略部分代码
}

新版本弃用了allowRequest()，取而代之的是attemptExecution()方法。正如方法名的含义，尝试执行。通过此方法的返回值决定执行正常逻辑，还是降级逻辑。

 @Override
public boolean attemptExecution() {if (properties.circuitBreakerForceOpen().get()) {return false;}if (properties.circuitBreakerForceClosed().get()) {return true;}if (circuitOpened.get() == -1) {return true;} else {if (isAfterSleepWindow()) {if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {//only the first request after sleep window should executereturn true;} else {return false;}} else {return false;}}
}

此方法的实现也比较直观

1. 如果circuitBreakerForceOpen=true，说明熔断器已经强制开启，所有请求都会被熔断。
2. 如果circuitBreakerForceClosed =true，说明熔断器已经强制关闭，所有请求都会被放行。
3. circuitOpened默认-1，用以保存最近一次发生熔断的时间戳。
   如果circuitOpened不等于 -1，说明已经发生熔断，通过isAfterSleepWindow()判断当前是否需要进行试探。判断逻辑是当前时间是否已经超过上次熔断的时间戳 + 试探窗口5000ms

// 睡眠窗口是否过去
private boolean isAfterSleepWindow() {final long circuitOpenTime = circuitOpened.get();final long currentTime = System.currentTimeMillis();final long sleepWindowTime = properties.circuitBreakerSleepWindowInMilliseconds().get();return currentTime > circuitOpenTime + sleepWindowTime;
}

4. 接下来进入if分支，通过compareAndSet修改变量status。只有睡眠窗口(5000ms)之后的第一个请求可以执行正常逻辑，且修改当前状态为HALF_OPEN，进入半熔断状态，其它请求执行compareAndSet(Status.OPEN, Status.HALF_OPEN)时都返回false，执行降级逻辑。

上述过程过程就是熔断器自动恢复的逻辑，总结一下:
当进入OPEN状态后，会进入一段睡眠窗口，即只会OPEN一段时间，所以这个睡眠窗口过去，就会从OPEN状态变成HALF_OPEN状态，这种设计是为了能做到弹性恢复，这种状态的变更，并不是由调度线程来做，而是由请求来触发。

5. HALF OPEN

HystrixCommand执行的过程中,当error的时候会触发circuitBreaker.markNonSuccess()，执行成功或者执行完成触发 circuitBreaker.markSuccess()。追溯源码，这两个方法的调用时发生在attemptExecution（）调用之后。在
AbstractCommand#executeCommandAndObserve中,下面省略的部分干扰代码。

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {final Action1<R> markEmits = new Action1<R>() {@Overridepublic void call(R r) {circuitBreaker.markSuccess();}};final Action0 markOnCompleted = new Action0() {@Overridepublic void call() {circuitBreaker.markSuccess();}};final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {@Overridepublic Observable<R> call(Throwable t) {circuitBreaker.markNonSuccess();}};return execution.doOnNext(markEmits).doOnCompleted(markOnCompleted).onErrorResumeNext(handleFallback).doOnEach(setRequestContext);
}

markSuccess中，若为半开状态时，会放第一个请求去执行，并跟踪它的执行结果，如果是成功，那么将由HALF_OPEN状态变成CLOSED状态，
markNonSuccess中，如果第一个被放去执行的请求执行失败（资源获取失败、异常、超时等），就会由HALP_OPEN状态再变为OPEN状态。

markNonSuccess(),通过compareAndSet修改status为熔断开启状态，并更新当前熔断开启的时间戳。

@Override
public void markNonSuccess() {if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {//This thread wins the race to re-open the circuit - it resets the start time for the sleep windowcircuitOpened.set(System.currentTimeMillis());}
}

markSuccess()

 @Override
public void markSuccess() {if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {//This thread wins the race to close the circuit - it resets the stream to start it over from 0metrics.resetStream();Subscription previousSubscription = activeSubscription.get();if (previousSubscription != null) {previousSubscription.unsubscribe();}Subscription newSubscription = subscribeToStream();activeSubscription.set(newSubscription);circuitOpened.set(-1L);}
}

如果断路器状态是半开并且成功设置为关闭状态时，执行以下步骤。1.重置度量指标对象,2.取消之前的订阅，发起新的订阅。3.设置断路器的打开时间为-1。

6. 滑动窗口原理

后面补充~

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！