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

Hystrix 的运行原理

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

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

断路器的工作原理

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

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

断路器源代码

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

/** 
 * 连接到 {@link HystrixCommand} 执行的断路器逻辑，如果失败超过定义的阈值，将停止允许执行。
 * 断路器会在执行 HystrixCommand 时调用断路器逻辑。如果故障超过定义的阈值，断路器熔断开关将打开，这将阻止任务执行。
 * 
 * 默认的（也是唯一的）实现将允许在定义的sleepWindow 之后进行一次重试，直到执行成功，此时它将再次关闭电路并允许再次执行
 * * 默认（且唯一）的实现将允许在定义的 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} 成功执行时调用，作为处于半开状态时的反馈机制的一部分。 
* 
* 当断路器处于半开状态时，作为反馈机制的一部分，从 HystrixCommand 的成功执行中调用。 
*/ 
void markSuccess(); 
/** 
* 在 {@link HystrixCommand} 执行不成功时调用，作为处于半开状态时的反馈机制的一部分。 
* 当断路器半开时，作为反馈机制的一部分，它会从 HystrixCommand 执行不成功的调用。 
*/ 
void markNonSuccess(); 
/** 
* 在命令执行开始时调用以尝试执行。这是非幂等的 - 它可能会修改内部状态。 
* 
* 在命令执行开始时调用尝试执行，主要使用的时间是判断请求是否可以执行。这不是幂等的 - 它可能会修改内部状态。 
*/
 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} 实例。
 * 
 * 这是线程安全的，并确保每个 {@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()方法有副作用，不能重复调用。