Hystrix原理和配置说明

Posted 2022-01-07 Leo Han

在微服务场景下，服务间调用链路加长，有时候一个部分出现问题可能会引起整个链路的崩溃，为此微服务中提出了如下两个概念：

• 服务熔断
• 服务降级

服务熔断

熔断来源于电路中断路器的概念，服务熔断指的是 当下游服务因为某种原因不可用或响应慢时，上游服务为了保证自己整体服务的可用性，不在调用下游服务而是直接返回，快速释放资源，直到下游服务恢复才继续调用

服务降级

当服务器压力剧增的时候,根据当前业务情况以及流量，对一些服务和页面有策略的降级，以此缓解服务器资源的压力以保障核心任务的正常运行，同时也保证了大部分客户能得到正常的响应。

服务熔断一般是针对调用的下游服务，而降级一般是从系统的整体考虑的

从官网找的Hystrix工作流图：

整体执行流程如下：

1. 构造一个 HystrixCommand或HystrixObservableCommand
2. 执行上面构造的command，会涉及到如下四个方法：

K             value   = command.execute();
Future<K>     fValue  = command.queue();
Observable<K> ohValue = command.observe();         //hot observable
Observable<K> ocValue = command.toObservable();    //cold observable

3. 如果调用允许缓存，判断请求的响应是否有缓存，如果有缓存，直接返回缓存内容
4. 判断断路器是否打开，如果断路器打开那么执行步骤8，断路器没有打开，执行步骤5
5. 如果线程池和队列或者信号量是否已经满了，如果已经满了，执行步骤8，没有达到阈值执行步骤6
6. 执行HystrixObservableCommand.construct()或HystrixCommand.run()
7. 统计计算断路器的状态，Hystrix会将请求的结果：成功、失败、拒绝、超时信息汇报给断路器，断路器根据这些信息是否打开断路器或关闭断路器
8. 当command执行失败的时候，执行Fallback
9. 当command执行成功，返回相应给调用者

断路器逻辑：

Hystrix是Netflix实现的服务限流、降级功能，可以完成隔离、限流、熔断、降级这些常用保护功能。Hystrix的隔离主要是基于线程隔离或者信号量隔离来实现的，服务隔离保证了不同服务之间的调用不会受其他服务影响。Hystrix的执行都是基于线程池异步调度执行的，而在方法调用上，同步方法实际也是异步执行的，只不过同步方法会阻塞等待，直到方法完成才返回。

基于Spring框架下，Hystrix基于AOP实现对方法的切面代理HystrixCommandAspect，获取方法是否有HystrixCollapser或HystrixCommand注解，如果方法上有这两个注解的话，那么Hystrix将在实际方法执行前执行Hystrix的逻辑。
Hystrix中通过HystrixCircuitBreaker来定义了断路器的相关逻辑，主要有两个实现：

• NoOpCircuitBreaker 不进行任何处理
• HystrixCircuitBreakerImpl Hystrix默认断路器实现

HystrixCircuitBreakerImpl主要就是基于HystrixCommandMetrics统计信息来设置当前断路器的状态：

class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker 
        private final HystrixCommandProperties properties;
        private final HystrixCommandMetrics metrics;
        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();
                
            
        

        @Override
        public 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;
        

        @Override
        public boolean isOpen() 
            if (circuitOpen.get()) 
                return true;
            
            HealthCounts health = metrics.getHealthCounts();
            if (health.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) 
                return false;
            

            if (health.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) 
                return false;
             else 
                if (circuitOpen.compareAndSet(false, true))              circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());
                    return true;
                 else 
                    return true;
                
            
        

    

这里通过allowRequest来判断断路器是否允许请求通过，而断路器的状态主要由一个AtomicBoolean类型保存，如果为true则表示当前断路器打开，而这里并没有维护一个半打开的状态，而是当状态为打开时，判断当前时间和断路器打开时间的间隔是否超过配置的断路器窗口时间，如果超过，那么这时候会将当前请求通过，同时通过CAS重置断路器状态打开的时间，注意这时候断路器状态依然是打开的，只不过通过时间的差异，放一个请求进去了，如果这个请求成功了，则会通过AtomicBoolean来关闭断路器，同时重置HystrixCommandMetrics信息。

`大家可以发现，Sentinel的断路器其实就是借鉴Hystrix的，打开、半打开、关闭状态的切换逻辑基本上就是一模一样的。

对于限流，Hystrix默认会为每个依赖服务开10个线程，如果10个线程都忙的话将拒绝执行任务，信号量同理

Hystrix为每个HystrixCommand中同样的grouoKey会分配同一个线程，Hysrix中线程池默认使用HystrixThreadPoolDefault实现，通过isQueueSpaceAvailable判断当前线程池是否能够继续添加任务。

public boolean isQueueSpaceAvailable() 
            if (queueSize <= 0) 
                return true;
             else 
                return threadPool.getQueue().size() < properties.queueSizeRejectionThreshold().get();
            
        

groupKey和threadPoolKey默认值是当前服务方法所在类的simpleName

在nETFLIX-Hystrix中，HystrixPropertiesManager定义相关的Hystrix的所有配置：

 /**
     * Command execution properties.
     */
    public static final String EXECUTION_ISOLATION_STRATEGY = "execution.isolation.strategy";
    public static final String EXECUTION_ISOLATION_THREAD_TIMEOUT_IN_MILLISECONDS = "execution.isolation.thread.timeoutInMilliseconds";
    public static final String EXECUTION_TIMEOUT_ENABLED = "execution.timeout.enabled";
    public static final String EXECUTION_ISOLATION_THREAD_INTERRUPT_ON_TIMEOUT = "execution.isolation.thread.interruptOnTimeout";
    public static final String EXECUTION_ISOLATION_SEMAPHORE_MAX_CONCURRENT_REQUESTS = "execution.isolation.semaphore.maxConcurrentRequests";

    /**
     * Command fallback properties.
     */
    public static final String FALLBACK_ISOLATION_SEMAPHORE_MAX_CONCURRENT_REQUESTS = "fallback.isolation.semaphore.maxConcurrentRequests";
    public static final String FALLBACK_ENABLED = "fallback.enabled";

    /**
     * Command circuit breaker properties.
     */
    public static final String CIRCUIT_BREAKER_ENABLED = "circuitBreaker.enabled";
    public static final String CIRCUIT_BREAKER_REQUEST_VOLUME_THRESHOLD = "circuitBreaker.requestVolumeThreshold";
    public static final String CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS = "circuitBreaker.sleepWindowInMilliseconds";
    public static final String CIRCUIT_BREAKER_ERROR_THRESHOLD_PERCENTAGE = "circuitBreaker.errorThresholdPercentage";
    public static final String CIRCUIT_BREAKER_FORCE_OPEN = "circuitBreaker.forceOpen";
    public static final String CIRCUIT_BREAKER_FORCE_CLOSED = "circuitBreaker.forceClosed";

    /**
     * Command metrics properties.
     */
    public static final String METRICS_ROLLING_PERCENTILE_ENABLED = "metrics.rollingPercentile.enabled";
    public static final String METRICS_ROLLING_PERCENTILE_TIME_IN_MILLISECONDS = "metrics.rollingPercentile.timeInMilliseconds";
    public static final String METRICS_ROLLING_PERCENTILE_NUM_BUCKETS = "metrics.rollingPercentile.numBuckets";
    public static final String METRICS_ROLLING_PERCENTILE_BUCKET_SIZE = "metrics.rollingPercentile.bucketSize";
    public static final String METRICS_HEALTH_SNAPSHOT_INTERVAL_IN_MILLISECONDS = "metrics.healthSnapshot.intervalInMilliseconds";

    /**
     * Command CommandRequest Context properties.
     */
    public static final String REQUEST_CACHE_ENABLED = "requestCache.enabled";
    public static final String REQUEST_LOG_ENABLED = "requestLog.enabled";

    /**
     * Thread pool properties.
     */
    public static final String MAX_QUEUE_SIZE = "maxQueueSize";
    public static final String CORE_SIZE = "coreSize";
    public static final String MAXIMUM_SIZE = "maximumSize";
    public static final String ALLOW_MAXIMUM_SIZE_TO_DIVERGE_FROM_CORE_SIZE = "allowMaximumSizeToDivergeFromCoreSize";
    public static final String KEEP_ALIVE_TIME_MINUTES = "keepAliveTimeMinutes";
    public static final String QUEUE_SIZE_REJECTION_THRESHOLD = "queueSizeRejectionThreshold";
    public static final String METRICS_ROLLING_STATS_NUM_BUCKETS = "metrics.rollingStats.numBuckets";
    public static final String METRICS_ROLLING_STATS_TIME_IN_MILLISECONDS = "metrics.rollingStats.timeInMilliseconds";

    /**
     * Collapser properties.
     */
    public static final String MAX_REQUESTS_IN_BATCH = "maxRequestsInBatch";
    public static final String TIMER_DELAY_IN_MILLISECONDS = "timerDelayInMilliseconds";

首先对于Command execution属性，主要是用例控制Hystrix执行方法时候的一些逻辑，主要如下：

#执行时隔离策略，主要有两个THREAD, SEMAPHORE，默认为THREAD,可以通过该`hystrix.command.default.`或者`hystrix.command.HystrixCommandKey.execution.isolation.strategy`来配置
execution.isolation.strategy 

#执行的超时时间，毫秒计算，可以通过`hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds`或者`hystrix.command.HystrixCommandKey.execution.isolation.thread.timeoutInMilliseconds`配置
execution.isolation.thread.timeoutInMilliseconds

#是否启用超时时间逻辑，默认为true
execution.timeout.enabled

#执行超时后，是否中断线程
execution.isolation.thread.interruptOnTimeout

#当使用信号量隔离的时候，该属性配置信号量的大小，用来控制并发，并发量超过该值时后续请求将被拒绝
execution.isolation.semaphore.maxConcurrentRequests

#该属性用来设置fallback执行的最大并发数，超过该值拒绝执行并抛出异常，信号量和线程池隔离都生效
fallback.isolation.semaphore.maxConcurrentRequests

#是否开启降级，如果为false那么当请求失败或拒绝的时候将不会执行falback逻辑降级
fallback.enabled

Command circuit breaker属性用来设置HystrixCircuitBreaker相关逻辑：

#当方法失败的时候是否使用断路器来跟踪相关指标和熔断，默认为true
circuitBreaker.enabled

#设置在滑动窗口中，断路器熔断的最小请求数，如设置了为10个，那么在一个滑动窗口（默认为10s）中收到了9个请求，即使这9个请求都失败了，断路器也不会打开
circuitBreaker.requestVolumeThreshold

#断路器打开状态的持续时间窗口，如果超过该窗口大小，那么断路器会进入半打开状态，这时候会放一个请求进来，如果请求成功，则会将断路器关闭，如果失败，断路器仍然打开，也就是断路器打开多久后尝试请求以期将断路器关闭
circuitBreaker.sleepWindowInMilliseconds

#用来设置断路器打开的错误请求占比，在请求超过circuitBreaker.requestVolumeThreshold阈值条件下，如果失败请求占比超过该值，那么断路器打开
circuitBreaker.errorThresholdPercentage

#强制打开断路器，这时候所有请求都会被拒绝，默认为false
circuitBreaker.forceOpen

#强制关闭断路器，这时候所有请求都会通过，circuitBreaker.forceOpen优先级比这个高
circuitBreaker.forceClosed

指标相关配置属性如下：

#用来设置对命令执行延迟是否使用百分比来跟踪和计算。默认为true，如果设置为false，那么所有的概要统计都将返回-1
metrics.rollingPercentile.enabled

#百分比统计的滑动窗口大小
metrics.rollingPercentile.timeInMilliseconds

#百分比统计中滑动窗口的桶的数量
metrics.rollingPercentile.numBuckets

#每个桶中最大执行数
metrics.rollingPercentile.bucketSize

#用来设置采集影响断路器状态的健康快照的间隔等待时间
metrics.healthSnapshot.intervalInMilliseconds

#该属性用来设置滚动时间窗的长度，单位为毫秒。该时间用于断路器判断健康度时需要收集信息的持续时间。断路器在收集指标信息时会根据设置的时间窗长度拆分成多个桶来累计各度量值，每个桶记录了一段时间的采集指标。例如，当为默认值10000毫秒时，断路器默认将其分成10个桶，每个桶记录1000毫秒内的指标信息
metrics.rollingStats.timeInMilliseconds

#用来设置滚动时间窗统计指标信息时划分“桶”的数量。默认值为10
metrics.rollingStats.numBuckets

request请求上线文相关配置：

#是否开启请求缓存
requestCache.enabled

#是否开启请求日志
requestLog.enabled

线程池相关配置：

#线程池核心线程数，默认为10
hystrix.threadpool.default.coreSize

#最大排队长度。默认-1，使用SynchronousQueue。其他值则使用 LinkedBlockingQueue。如果要从-1换成其他值则需重启，无法动态调整
hystrix.threadpool.default.maxQueueSize

#线程池排队数量阈值，如果达到该阈值，后续请求将拒绝，hystrix.threadpool.default.maxQueueSize=-1时，该配置不生效
hystrix.threadpool.default.queueSizeRejectionThreshold

#允许线程池的最大数量，也是能够支持的最大并发，但是必须咋设置了allowMaximumSizeToDivergeFromCoreSize属性才能生效
hystrix.threadpool.default.maximumSize

#当coreSize < maximumSize时，该属性控制一个线程多久不用后被释放，minute计数，默认为1分钟
hystrix.threadpool.default.keepAliveTimeMinutes

#该配置使maximumSize生效，默认情况下maximumSize=coreSize，，启用该配置后，coreSize < maximumSize，能够支持最大的并发为maximumSize
hystrix.threadpool.default.allowMaximumSizeToDivergeFromCoreSize

配置相关的，大家可以查下github上Hystrix的wiki: https://github.com/Netflix/Hystrix/wiki/Configuration 这里详细描述了相关的配置的作用。