Hystrix实现熔断降级

一、Hystrix执行过程

二、Hystrix熔断机制

1. 熔断原理

2. HystrixCircuitBreakerImpl源码解析

3. 熔断器开关状态

4. 采样统计

三、配置参数

四、参考资料

一、Hystrix执行过程

step1：每次调用创建一个新的HystrixCommand，把依赖调用封装在run()方法中。
step2：执行execute()/queue做同步或异步调用。
step3：判断熔断器(circuit-breaker)是否打开，如果打开跳到步骤8，进行降级策略，如果关闭进入步骤。
step4：判断线程池/队列/信号量是否跑满，如果跑满进入降级步骤8，否则继续后续步骤。
step5：调用HystrixCommand的run方法。运行依赖逻辑
step5a：依赖逻辑调用超时，进入步骤8。
step6：判断逻辑是否调用成功
step6a：返回成功调用结果
step6b：调用出错，进入步骤8。
step7：计算熔断器状态，所有的运行状态(成功, 失败, 拒绝,超时)上报给熔断器，用于统计从而判断熔断器状态。
step8：getFallback()降级逻辑。
以下四种情况将触发getFallback调用：
(1)：run()方法抛出非HystrixBadRequestException异常
(2)：run()方法调用超时
(3)：熔断器开启拦截调用
(4)：线程池/队列/信号量是否跑满
step8a：没有实现getFallback的Command将直接抛出异常
step8b：fallback降级逻辑调用成功直接返回
step8c：降级逻辑调用失败抛出异常
step9：返回执行成功结果

二、Hystrix熔断机制

1. 熔断原理

如上图是熔断器的类图，默认类为HystrixCircuitBreakerImpl。以上两个类都是HystrixCircuitBreaker接口的内部静态类。其中HystrixCircuitBreaker.HystrixCircuitBreakerImpl#allowRequest()方法判定是否熔断。如下图所示，若为true，则没有熔断，进入Command#run()；false，则进入熔断，Command#进入getFallback()。

2. HystrixCircuitBreakerImpl源码解析

方法名称	作用
boolean allowRequest()	判断是否熔断
boolean allowSingleTest()	允许一段时间窗口内测试服务器是否正常，若成功熔断开关关闭
boolean isOpen()	当前熔断器是否进行熔断（根据统计）
void markSuccess()	熔断器开关关闭后，重置统计数据

package com.netflix.hystrix;import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;import com.netflix.hystrix.HystrixCommandMetrics.HealthCounts;public interface HystrixCircuitBreaker {....../* package */static class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {// 参数配置private final HystrixCommandProperties properties;// 采样统计private final HystrixCommandMetrics metrics;// 熔断器状态：打开或关闭，默认为false，即关闭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 (circuitOpen.get()) {// 尝试关闭熔断器成功if (circuitOpen.compareAndSet(true, false)) {// 重置采样统计metrics.resetStream();}}}// 判断请求是否熔断：true,run();false,getFallback()@Overridepublic boolean allowRequest() {// 熔断开关强制打开，则进入熔断if (properties.circuitBreakerForceOpen().get()) {return false;}// 熔断开关强制关闭，则正常处理if (properties.circuitBreakerForceClosed().get()) {// 处理当前请求，采样统计后，熔断器是否需要打开isOpen();return true;}return !isOpen() || allowSingleTest();}// 一段时间窗口内允许请求（半打开状态）—— 自动恢复public boolean allowSingleTest() {// 熔断器打开时的最后一个测试请求的时间戳long timeCircuitOpenedOrWasLastTested = circuitOpenedOrLastTestedTime.get();// 熔断器打开// 当前时间戳 > 熔断器打开时的最后一个测试请求的时间戳 + 一段时间窗口内（配置获取）if (circuitOpen.get() && System.currentTimeMillis() > timeCircuitOpenedOrWasLastTested + properties.circuitBreakerSleepWindowInMilliseconds().get()) {// 尝试更新最后一个测试请求的时间戳成功if (circuitOpenedOrLastTestedTime.compareAndSet(timeCircuitOpenedOrWasLastTested, System.currentTimeMillis())) {// 一段时间窗口内允许请求return true;}}return false;}// 判定当前熔断器是否进行熔断（根据统计）@Overridepublic boolean isOpen() {// 熔断器已经打开if (circuitOpen.get()) {return true;}// 熔断器开关关闭，则根据采样是否需要打开熔断// 健康统计 —— 采样统计类HealthCounts health = metrics.getHealthCounts();// 采样总请求数 < 配置circuitBreakerRequestVolumeThreshold阈值，不熔断if (health.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {return false;}// 采样错误率 < 配置circuitBreakerErrorThresholdPercentage阈值，不熔断if (health.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {return false;// 采样错误率 > 配置circuitBreakerErrorThresholdPercentage阈值，熔断} else {// 尝试打开熔断器成功if (circuitOpen.compareAndSet(false, true)) {// 设置熔断器打开时的最后一个测试请求的时间戳为当前时间戳circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());return true;} else {// 失败时，也进行熔断return true;}}}}......}

3. 熔断器开关状态

闭合_Closed：强制关闭或闭合状态，不启动熔断，即：不降级处理。

打开_Open：强制打开或打开状态，启动熔断，调用getFallback()。

半打开_Half-Open：打开状态，一段时间窗口内进行重试，成功后开关关闭；否则还处于打开。

4. 采样统计

BucketedCounterStream：计数统计：记录一段时间窗口内的失败、超时、线程拒绝，成功（为一组）。统计时采用N-1组数据统计，第N组刚开始统计时随时间变化。基于时间转滚统计。

RollingConcurrencyStream：最大并发数统计：线程池最大并发数。

RollingDistributionStream：延时百分比统计：记录一段时间窗口内的百分位统计，对N-1组百分比数据排序，P50、P99、P999。

package com.netflix.hystrix;import com.netflix.hystrix.metric.HystrixCommandCompletion;
import com.netflix.hystrix.metric.HystrixThreadEventStream;
import com.netflix.hystrix.metric.consumer.CumulativeCommandEventCounterStream;
import com.netflix.hystrix.metric.consumer.HealthCountsStream;
import com.netflix.hystrix.metric.consumer.RollingCommandEventCounterStream;
import com.netflix.hystrix.metric.consumer.RollingCommandLatencyDistributionStream;
import com.netflix.hystrix.metric.consumer.RollingCommandMaxConcurrencyStream;
import com.netflix.hystrix.metric.consumer.RollingCommandUserLatencyDistributionStream;
import com.netflix.hystrix.strategy.HystrixPlugins;
import com.netflix.hystrix.strategy.eventnotifier.HystrixEventNotifier;
import com.netflix.hystrix.util.HystrixRollingNumberEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.functions.Func0;
import rx.functions.Func2;import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;public class HystrixCommandMetrics extends HystrixMetrics {......public static class HealthCounts {// 请求总数计数private final long totalCount;// 请求失败计数 = 失败 + 超时 + 线程池拒绝 + 信号量拒绝private final long errorCount;// 请求失败错误率private final int errorPercentage;HealthCounts(long total, long error) {this.totalCount = total;this.errorCount = error;if (totalCount > 0) {this.errorPercentage = (int) ((double) errorCount / totalCount * 100);} else {this.errorPercentage = 0;}}private static final HealthCounts EMPTY = new HealthCounts(0, 0);public long getTotalRequests() {return totalCount;}public long getErrorCount() {return errorCount;}public int getErrorPercentage() {return errorPercentage;}// 统计错误数量public HealthCounts plus(long[] eventTypeCounts) {long updatedTotalCount = totalCount;long updatedErrorCount = errorCount;// 成功long successCount = eventTypeCounts[HystrixEventType.SUCCESS.ordinal()];// 失败long failureCount = eventTypeCounts[HystrixEventType.FAILURE.ordinal()];// 超时long timeoutCount = eventTypeCounts[HystrixEventType.TIMEOUT.ordinal()];// 线程池拒绝long threadPoolRejectedCount = eventTypeCounts[HystrixEventType.THREAD_POOL_REJECTED.ordinal()];// 信号量拒绝long semaphoreRejectedCount = eventTypeCounts[HystrixEventType.SEMAPHORE_REJECTED.ordinal()];// 总计 = 成功 + 失败 + 超时 + 线程池拒绝 + 信号量拒绝updatedTotalCount += (successCount + failureCount + timeoutCount + threadPoolRejectedCount + semaphoreRejectedCount);// 错误 = 失败 + 超时 + 线程池拒绝 + 信号量拒绝updatedErrorCount += (failureCount + timeoutCount + threadPoolRejectedCount + semaphoreRejectedCount);return new HealthCounts(updatedTotalCount, updatedErrorCount);}public static HealthCounts empty() {return EMPTY;}public String toString() {return "HealthCounts[" + errorCount + " / " + totalCount + " : " + getErrorPercentage() + "%]";}}
}

三、配置参数

hystrix.command.[commandkey].execution.isolation.strategy：隔离策略THREAD（默认）或SEMAPHORE。
hystrix.command.[commandkey].execution.timeout.enabled：是否开启超时设置，默认true。
hystrix.command.[commandkey].execution.isolation.thread.timeoutInMilliseconds：默认超时时间，默认1000ms。
hystrix.command.[commandkey].execution.isolation.thread.interruptOnTimeout：是否打开超时线程中断，默认值true。
hystrix.command.[commandkey].execution.isolation.thread.interruptOnFutureCancel：当隔离策略为THREAD时，当执行线程执行超时时，是否进行中断处理，即Future#cancel(true)处理，默认为false。
hystrix.command.[commandkey].execution.isolation.semaphore.maxConcurrentRequests：信号量最大并发度默认值10，该参数当使用ExecutionIsolationStrategy.SEMAPHORE策略时才有效。如果达到最大并发请求数，请求会被拒绝。理论上选择semaphore size的原则和选择thread size一致，但选用semaphore时每次执行的单元要比较小且执行速度快（ms级别），否则的话应该用thread。
hystrix.command.[commandkey].fallback.isolation.semaphore.maxConcurrentRequests：fallback方法的信号量配置，配置getFallback方法并发请求的信号量，如果请求超过了并发信号量限制，则不再尝试调用getFallback方法，而是快速失败，默认信号量为10。
hystrix.command.[commandkey].fallback.enabled：是否启用降级处理，如果启用了，则在超时或异常时调用getFallback进行降级处理，默认开启。

hystrix.command.[commandkey].circuitBreaker.enabled：是否开启熔断机制，默认为true。
hystrix.command.[commandkey].circuitBreaker.forceOpen：强制开启熔断，默认为false。
hystrix.command.[commandkey].circuitBreaker.forceClosed：强制关闭熔断，默认为false。
hystrix.command.[commandkey].circuitBreaker.sleepWindowInMilliseconds：熔断窗口时间，默认为5s。
hystrix.command.[commandkey].circuitBreaker.requestVolumeThreshold：当在配置时间窗口内达到此数量后的失败，进行短路。默认20个。
hystrix.command.[commandkey].circuitBreaker.errorThresholdPercentage：出错百分比阈值，当达到此阈值后，开始短路。默认50%。

四、参考资料

Hystrix实现线程隔离_爱我所爱0505-CSDN博客

微服务高可用利器——Hystrix熔断降级原理&实践总结_舒哥的blog-CSDN博客_hystrix熔断原理

熔断机制hystrix - 佳716 - 博客园

服务熔断、线程池和信号量隔离、Feign服务降级项目、数据监控dashboard、Turbine聚合 - 知乎

hystrix熔断器之配置 - zwh1988 - 博客园