Hystrix 原理深入分析-spring cloud 入门教程

Hystrix 的运行原理

1. 构造一个 HystrixCommand 或 HystrixObservableCommand 对象
2. 执行命令。
3. 检查缓存是否被命中，如果命中则直接返回。
4. 检查断路器开关是否断开。如果是开路，则直接熔断，经过回退逻辑。
5. 检查线程池/队列/信号量是否已满。如果线程池/队列/信号量已满，则直接拒绝请求并遵循回退逻辑。
6. 如果不满足上述条件，则调用 HYST rixObservableCommand.construct() 方法 HystrixCommand.run Method() 执行业务逻辑。
7. 判断业务逻辑方法运行是否有异常或超时。如果是这样，它将直接降级并使用回退逻辑。
8. 上报统计数据，由用户计算断路器状态。
9. 返回结果

从流程图中可以发现错误统计只有在5和7的情况下才会上报。

断路器的工作原理

断路器的开关控制逻辑如下：

1. 在一个统计时间窗口（HYSTrixCommandProperties.metricsRollingStatisticalWindowInMilliseconds()）内，处理的请求数达到设置的最小阈值（HYST）rixCommandProperties.circuitBreakerRequestVolumeThreshold（）），错误百分比超过设置的最大阈值（HYSTrixCommandProperties.circuitBreakerThreshold（） ) )此时断路器分闸，断路器状态由合闸切换为分闸。
2. 当断路器断开时，它将直接融断所有请求（快速失败）并经过回退逻辑。
3. 经过一个休眠窗口时间（HYST rixCommandProperties.circuitBreakerSleepWindowInMilliseconds()），Hystrix会释放一个进行后续服务并将断路器开关切换到半开（half OPEN）。如果请求失败，断路器将熔断开关切换到OPEN状态，继续熔断所有请求，直到下一个休眠时间窗口到来；如果请求成功，断路器将切换到 CLOSED 状态，此时允许所有请求通过，直到发生一步，断路器开关才会切换到 OPEN 状态。

断路器源代码

Hystrix 断路器的实现类是 HystrixCircuitBreaker。源代码如下：

/** 
 * 连接到 {@link HystrixCommand} 执行的断路器逻辑，如果失败超过定义的阈值，将停止允许执行。
 * 断路器会在执行 HystrixCommand 时调用断路器逻辑。如果故障超过定义的阈值，断路器熔断开关将打开，这将阻止任务执行。
 * <p> 
 * 默认的（也是唯一的）实现将允许在定义的sleepWindow 之后进行一次重试，直到执行成功，此时它将再次关闭电路并允许再次执行
 * <p> * 默认（且唯一）的实现将允许在定义的 sleepWindow 之后进行一次重试，直到成功执行，此时它将再次关闭电路并允许再次执行。 
*/  
public interface HystrixCircuitBreaker {  

/**  
 * 每个 {@link HystrixCommand} 请求都会询问是否允许继续。没有副作用并且是幂等，不修改任何内部状态，并考虑了半开逻​​辑 

 * 每个HystrixCommand 请求询问是否允许继续。 它是幂等的，不修改任何内部状态。考虑半开放逻辑，当一个sleep window到来时，会释放一些请求给后续逻辑 
 * @return boolean 是否允许请求（是否允许请求） 

 */  

boolean allowRequest();  

/**  
* 断路器当前是否打开（跳闸）。 
* 判断熔断器开关是否为OPEN（如果是OPEN或者half_OPEN时返回true。如果是CLOSE则返回false。没有副作用，是幂等的）。 
* @return 断路器的布尔状态（返回断路器状态） 
*/  
boolean isOpen();   
/**  
* 在 {@link HystrixCommand} 成功执行时调用，作为处于半开状态时的反馈机制的一部分。 
* <p>  
* 当断路器处于半开状态时，作为反馈机制的一部分，从 HystrixCommand 的成功执行中调用。 
*/  
void markSuccess();   
/** 
* 在 {@link HystrixCommand} 执行不成功时调用，作为处于半开状态时的反馈机制的一部分。 
* 当断路器半开时，作为反馈机制的一部分，它会从 HystrixCommand 执行不成功的调用。 
*/  
void markNonSuccess();   
/**  
* 在命令执行开始时调用以尝试执行。这是非幂等的 - 它可能会修改内部状态。 
* <p>  
* 在命令执行开始时调用尝试执行，主要使用的时间是判断请求是否可以执行。这不是幂等的 - 它可能会修改内部状态。 
*/

   boolean attemptExecution();

 }

断路器的默认实现是它的内部类 /** * @ExcludeFromJavadoc * @ThreadSafe */ class Factory { // String类型的HystrixCommandKey.name()（我们不能直接使用 HystrixCommandKey，因为我们不能保证它正确实现了 hashcode/equals） private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();

/** * 根据 HystrixCommandKey获取HystrixCircuitBreaker

* 获取给定 {@link HystrixCommandKey} 的 {@link HystrixCircuitBreaker} 实例。 * <p> * 这是线程安全的，并确保每个 {@link HystrixCommandKey} 只有 1 个 {@link HystrixCircuitBreaker}。 * * {@link HystrixCommand} 实例的 * @param key {@link HystrixCommandKey} 请求 {@link HystrixCircuitBreaker} * @param group Pass-thru to {@link HystrixCircuitBreaker} * @param properties Pass-thru to {@link HystrixCircuitBreaker} * @param metrics 传递到 {@link HystrixCircuitBreaker} * @return {@link HystrixCircuitBreaker} for {@link HystrixCommandKey} */ public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) { // 根据 HystrixCommandKey 获取断路器 HystrixCircuitBreaker previousCached = circuitBreakersByCommand.get(key.name()); if (previouslyCached != null) { return previousCached; }

// 如果我们到达这里，这是第一次，所以我们需要初始化

// 创建并添加到映射中...使用 putIfAbsent 原子地处理 // 2个线程同时到达该点的可能会竞争，所以采用ConcurrentHashMap 为我们提供线程安全 // 如果 2 个线程在这里命中，则只会添加一个线程，而另一个将获得非空响应。 // 第一次没有拿到断路器，需要初始化 // 这里直接使用concurrenchashmap的putIfAbsent方法。这是一个原子操作。如果这里添加了两个线程执行，那么只有一个线程会将值放入容器中 // 让我们保存锁定步骤 HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics) ); if (cbForCommand == null) { // 这意味着 putIfAbsent 步骤刚刚创建了一个新的实例，所以让我们再次检索并返回它 return circuitBreakersByCommand.get(key.name()); } else { // 这意味着发生了竞争，并且在尝试“放置”另一个之前到达那里时 // 我们取而代之的是检索它，现在将返回它 return cbForCommand; } }

/** * 根据HystrixCommandKey获取HystrixCircuitBreaker。如果它不返回 NULL * 获取给定 {@link HystrixCommandKey} 的 {@link HystrixCircuitBreaker} 实例，如果不存在，则为 null。 * * {@link HystrixCommand} 实例的 @param key {@link HystrixCommandKey} 请求 {@link HystrixCircuitBreaker} * @return {@link HystrixCircuitBreaker} 为 {@link HystrixCommandKey} */ public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) { return circuitBreakersByCommand.get(key.name()); }

/** * 清除所有断路器。如果新请求进来，实例将被重新创建。 * 清除所有断路器。如果有新的请求，断路器将重新创建并放置在容器中。 */ static void reset() { circuitBreakersByCommand.clear(); } } /** * 默认断路器实现 * {@link HystrixCircuitBreaker} 的默认生产实现。 * * @ExcludeFromJavadoc * @ThreadSafe */ /* package */ class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker { private final HystrixCommandProperties properties; 私有的最终 HystrixCommandMetrics 指标；

enum Status { // 断路器状态，闭合，断开，半开 CLOSED, OPEN, HALF_OPEN; }

// 赋值不是线程安全的。如果想实现不加锁，可以使用atomicreference<v>来更新对象引用的atom。 // AtomicReference原子引用保证Status的原子性修改 private final AtomicReference<Status> status = new AtomicReference<Status>(Status.CLOSED); // 记录断路器分闸的时间点（时间戳）。如果时间大于0，则表示断路器打开或半开 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;

//在定时器上，这将在命令执行发生时设置开/关之间的电路

Subscription s = subscribeToStream(); activeSubscription.set(s); }

private Subscription subscribeToStream() { /* * 此流将重新计算健康流中每个 onNext 的 OPEN/CLOSED 状态 */ return metrics.getHealthCountsStream() .observe() .subscribe(new Subscriber<HealthCounts>() { @Override public void onCompleted() {

}

@Override public void onError(Throwable e) {

}

@Override public void onNext(HealthCounts hc) { // check if we are past the statisticalWindowVolumeThreshold // Check the minimum number of requests in a time window //检查是否是超过statisticalWindowVolumeThreshold值 //检查时间窗口请求的最小数目 if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {

// 当没有超过统计窗口的最小量阈值时，断路器的状态没有变化。 // 如果它原来是被关闭，它保持关闭 // 如果它是半开的，我们需要等待一个成功的命令执行 // 如果它被打开，我们需要等待睡眠窗口过去

} else {

// 检查错误比例阈值 if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) { //we are not past the minimum error threshold for the stat window, // so no change to circuit status. // if it was CLOSED, it stays CLOSED // if it was half-open, we need to wait for a successful command execution // if it was open, we need to wait for sleep window to elapse

// 当没有超过统计窗口的最小错误阈值时，电路状态没有变化。 // 如果它是 CLOSED，它保持 CLOSED // 如果它是半开的，我们需要等待一个成功的命令执行 // 如果它是开放的，我们需要等待睡眠窗口过去 } else { // 我们的失败率太高，我们需要将状态设置为 OPEN

if (status.compareAndSet(Status.CLOSED, Status.OPEN)) { circuitOpened.set(System.currentTimeMillis()); } } } } }); }

@Override public void markSuccess() { // The circuit breaker is processing half open and the HystrixCommand is executed successfully. Set the status to off //断路器正在处理半开和HystrixCommand被成功执行。将状态设置为关闭 if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {

//这个线程获得关闭电路的权限——它重置了流以从0重新开始

metrics.resetStream(); Subscription previousSubscription = activeSubscription.get(); if (previousSubscription != null) { previousSubscription.unsubscribe(); } Subscription newSubscription = subscribeToStream(); activeSubscription.set(newSubscription); circuitOpened.set(-1L); } }

@Override public void markNonSuccess() { //该断路器是半开和HystrixCommand被成功执行。将状态设置为打开 if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) { //此线程赢得重新打开电路的竞赛 - 它重置睡眠窗口的开始时间 circuitOpened.set(System.currentTimeMillis()); } }

@Override public boolean isOpen() { // 获取配置判断断路器是否处于强制断开的状态 if (properties.circuitBreakerForceOpen().get()) { return true; } // 获取配置判断断路器是否强制闭合的状态 if (properties.circuitBreakerForceClosed().get()) { return false; } return circuitOpened.get() >= 0; }

@Override public boolean allowRequest() {

//获取配置来判断断路器是否被强制打开 if (properties.circuitBreakerForceOpen().get()) { return false; } // Obtain the configuration to judge whether the circuit breaker is forced to close // 获取配置判断断路器是否强制闭合的 if (properties.circuitBreakerForceClosed().get()) { return true; } if (circuitOpened.get() == -1) { return true; } else { // If it is half open, the return does not allow Command execution // 如果是半开，则返回不允许命令执行 if (status.get().equals(Status.HALF_OPEN)) { return false; } else { // Check if the sleep window is over // 检查睡眠窗口是否结束 return isAfterSleepWindow(); } } }

private boolean isAfterSleepWindow() { final long circuitOpenTime = circuitOpened.get(); final long currentTime = System.currentTimeMillis(); // Gets the configured time window for sleep // 获取睡眠窗口配置的时间 final long sleepWindowTime = properties.circuitBreakerSleepWindowInMilliseconds().get(); return currentTime > circuitOpenTime + sleepWindowTime; }

@Override public boolean attemptExecution() { // Obtain the configuration to judge whether the circuit breaker is forced to open //获取配置来判断断路器是否处于被强制打开的状态 if (properties.circuitBreakerForceOpen().get()) { return false; } // Obtain the configuration to judge whether the circuit breaker is forced to close // 获取判断断路器是否强制合闸的配置 if (properties.circuitBreakerForceClosed().get()) { return true; } if (circuitOpened.get() == -1) { return true; } else { if (isAfterSleepWindow()) {

//只有在睡眠窗口时间过后的第一个请求才应该执行 //如果执行命令成功，状态将转换为CLOSED //如果执行命令失败，状态将转换为OPEN //如果正在执行的命令被取消订阅，状态将转换为 OPEN

if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) { return true; } else { return false; } } else { return false; } } }

}

ispen()：判断熔断器开关是否为OPEN（如果是OPEN或者half_OPEN则返回true，如果是CLOSE则返回false。没有副作用，是幂等的）。 allowRequest()：每个HystrixCommand请求询问是否允许继续（当断路器开关闭合或下一个睡眠窗口返回true时），它是幂等的，不修改任何内部状态. 考虑到半开放的逻辑，当一个sleep window到来时，它会释放一些请求给后续的逻辑。 attemptExecution()：在命令执行开始时调用以尝试执行。主要是用时间来判断请求是否可以执行。这是非幂等的，可能会修改内部状态。需要注意的是，isOpen()和allowRequest()方法是幂等的，可以重复调用；attemptExecution()方法有副作用，不能重复调用。