Spring Cloud 源码学习之 Hystrix 熔断器

 2019-11-23 11:09  阅读(1217)
文章分类:Spring Cloud

欢迎访问陈同学博客原文

文中源码基于 Spring Cloud Finchley.SR1、Spring Boot 2.0.6.RELEASE.

本文学习了Hystrix熔断器的原理、配置和源码,包含滑动窗口、状态变化等。

简介

circuit-breaker: circuit表示电路,大家译为熔断器非常精准。

回想起小时候,家里保险丝突然被烧断,需 手工更换一根新的保险丝;后来,保险丝被取代,电流过大时会跳闸,闸拉上去后立马恢复供电;等到上大学时,只要打开功率高一点的电吹风,砰的一声就断电,但过10分钟就自动来电。在电流过大时,通过熔断机制以保护电路和家电。

Hystrix 属于上面的第三种,一种自动恢复的智能熔断器,区别在于它保护的是系统,且判断 “电流过大” 的方式是:不断收集请求指标信息(sucess、failure、timeout、rejection),当达到设定熔断条件时(默认是请求失败率达到50%)进行熔断。

Spring Cloud 源码学习之 Hystrix Metrics 收集 一文中,学习了 Metrics 收集,这是上文的图。

20191123100240\_1.png

Hystrix Command 执行过程中,各种情况都以事件形式发出,再封装成特定的数据结构,最后汇入到事件流中(HystrixEventStream)。事件流提供了 observe() 方法,摇身一变,事件流把自己变成了一个数据源(各小溪汇入成河,消费者从河里取水),其他消费者可以从这里获取数据,而 circuit-breaker 就是消费者之一。

原理

在统计中,会使用一定数量的样本,并将样本进行分组,最后进行统计分析。

Hystrix 有点类似,例如:以秒为单位来统计请求的处理情况(成功请求数量、失败请求数、超时请求数、被拒绝的请求数),然后每次取最近10秒的数据来进行计算,如果失败率超过50%,就进行熔断,不再处理任何请求

这是Hystrix官网的一张图:

20191123100240\_2.png

它演示了 Hystrix 滑动窗口 策略,假定以秒为单位来统计请求处理情况,上面每个格子代表1秒,格子中的数据就是1秒内各处理结果的请求数量,格子称为 Bucket(译为桶)。

若每次的决策都以10个Bucket的数据为依据,计算10个Bucket的请求处理情况,当失败率超过50%时就熔断。10个Bucket就是10秒,这个10秒就是一个 滑动窗口(Rolling window)

为什么叫滑动窗口?因为在没有熔断时,每当收集好一个新的Bucket后,就会丢弃掉最旧的一个Bucket。上图中的深色的(23 5 2 0)就是被丢弃的桶,这和拿着放大镜从左到右看书有点类似,视野永远是放大镜那一部分。

下面是官方完整的流程图,策略是:不断收集数据,达到条件就熔断;熔断后拒绝所有请求一段时间(sleepWindow);然后放一个请求过去,如果请求成功,则关闭熔断器,否则继续打开熔断器。

20191123100240\_3.png

相关配置

默认配置都在HystrixCommandProperties类中。

先看两个metrics收集的配置。

  • metrics.rollingStats.timeInMilliseconds

表示滑动窗口的时间(the duration of the statistical rolling window),默认10000(10s),也是熔断器计算的基本单位。

  • metrics.rollingStats.numBuckets

滑动窗口的Bucket数量(the number of buckets the rolling statistical window is divided into),默认10. 通过timeInMilliseconds和numBuckets可以计算出每个Bucket的时长。

metrics.rollingStats.timeInMilliseconds % metrics.rollingStats.numBuckets 必须等于 0,否则将抛异常。

再看看熔断器的配置。

  • circuitBreaker.requestVolumeThreshold

滑动窗口触发熔断的最小请求数。如果值是20,但滑动窗口的时间内请求数只有19,那即使19个请求全部失败,也不会熔断,必须达到这个值才行,否则样本太少,没有意义。

  • circuitBreaker.sleepWindowInMilliseconds

这个和熔断器自动恢复有关,为了检测后端服务是否恢复,可以放一个请求过去试探一下。sleepWindow指的发生熔断后,必须隔sleepWindow这么长的时间,才能放请求过去试探下服务是否恢复。默认是5s

  • circuitBreaker.errorThresholdPercentage

错误率阈值,表示达到熔断的条件。比如默认的50%,当一个滑动窗口内,失败率达到50%时就会触发熔断。

源码学习

HystrixCircuitBreaker的创建

circuitBreaker是AbstractCommand的成员变量,AbstractCommand是HystrixCommand和HystrixObservableCommand的父类,因此每个command都有个circuitBreaker属性。

abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {
        protected final HystrixCircuitBreaker circuitBreaker;
    }

在AbstractCommand构造器中初始化circuitBreaker。

private static HystrixCircuitBreaker initCircuitBreaker(boolean enabled, HystrixCircuitBreaker fromConstructor, HystrixCommandKey commandKey...) {
        // 如果启用了熔断器
        if (enabled) {
            // 若commandKey没有对应的CircuitBreaker,则创建
            if (fromConstructor == null) {
                return HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics);
            } else {
                // 如果有则返回现有的
                return fromConstructor;
            }
        } else {
            return new NoOpCircuitBreaker();
        }
    }

再看看 HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics) 如何创建circuit-breakder?

circuitBreaker以commandKey为维度,每个commandKey都会有对应的circuitBreaker。

public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
        // 如果有则返回现有的, key.name()即command的name作为检索条件
        HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());
        if (previouslyCached != null) {
            return previouslyCached;
        }

        // 如果没有则创建并cache
        HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics));
        if (cbForCommand == null) {
            return circuitBreakersByCommand.get(key.name());
        } else {
            return cbForCommand;
        }
    }

如何订阅HystrixEventStream

本文最前面说到,HystrixEventStream提供了结构化的数据,提供了一个Observable对象,Hystrix只需要订阅它即可。

这HystrixCircuitBreaker接口实现类的构造器:

protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
        this.properties = properties;
        // 这是Command中的metrics对象,metrics对象也是commandKey维度的
        this.metrics = metrics;
        // !!!重点:订阅事件流
        Subscription s = subscribeToStream();
        activeSubscription.set(s);
    }

    // 订阅事件流, 前面打的比方: 小溪汇成的河, 各事件以结构化数据汇入了Stream中
    private Subscription subscribeToStream() {
        // HealthCountsStream是重点,下面会分析
        return metrics.getHealthCountsStream()
                .observe()
                // 利用数据统计的结果HealthCounts, 实现熔断器
                .subscribe(new Subscriber<HealthCounts>() {
                    @Override
                    public void onCompleted() {}

                    @Override
                    public void onError(Throwable e) {}

                    @Override
                    public void onNext(HealthCounts hc) {
                        // 检查是否达到最小请求数,默认20个; 未达到的话即使请求全部失败也不会熔断
                        if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
                            // 啥也不做
                        } else {
                            // 错误百分比未达到设定的阀值
                            if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
                            } else {
                                // 错误率过高, 进行熔断
                                if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
                                    circuitOpened.set(System.currentTimeMillis());
                                }
                            }
                        }
                    }
                });
    }

HealthCounts 属性如下,表示一个滑动窗口内的统计数据。

public static class HealthCounts {
        // rolling window 中请求总数量
        private final long totalCount;
        // 错误请求数(failure + success + timeout + threadPoolRejected + semaphoreRejected)
        private final long errorCount;
        // 错误率
        private final int errorPercentage;
    }

滑动窗口的实现

跟进下 metrics.getHealthCountsStream().observe(),那熔断器是如何进行数据统计的?

首先看下HystrixCommandMetrics的构造器,会初始化healthCountsStream这个健康统计数据流。

private HealthCountsStream healthCountsStream;
    HystrixCommandMetrics(final HystrixCommandKey key, ...) {
        // 又是以key为维度
        healthCountsStream = HealthCountsStream.getInstance(key, properties);
    }

对于HealthCountsStream的部分注释如下,翻译过来不够准确,简单中英对照下:

Maintains a stream of rolling health counts for a given Command.
    它是一份流数据, 承载了指定Command的健康统计数据

    There is a rolling window abstraction on this stream.
    基于这个stream, 抽象出了滑动窗口

    The HealthCounts object is calculated over a window of t1 milliseconds.  This window has b buckets.
    HealthCounts的数据是根据一个t1毫秒的滑动窗口计算得来,这个窗口有b个Buckets

    Therefore, a new HealthCounts object is produced every t2 (=t1/b) milliseconds
    因此, 每 t2 (=t1/b)毫秒就会产生一个HealthCounts作为统计结果

HealthCountsStream.getInstance如下:

public static HealthCountsStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties) {
        // 每个Bucket的时间长度
        final int healthCountBucketSizeInMs = properties.metricsHealthSnapshotIntervalInMilliseconds().get();
        // 滑动窗口的时间/每个Bucket的时间长度=滑动窗口内Bucket的数量
        final int numHealthCountBuckets = properties.metricsRollingStatisticalWindowInMilliseconds().get() / healthCountBucketSizeInMs;

        return getInstance(commandKey, numHealthCountBuckets, healthCountBucketSizeInMs);
    }

    // 以Key为维度,每个Key有自己唯一的一个HealthCountsStream
    public static HealthCountsStream getInstance(HystrixCommandKey commandKey, int numBuckets, int bucketSizeInMs) {
        HealthCountsStream initialStream = streams.get(commandKey.name());
        if (initialStream != null) {
            return initialStream;
        } else {
            final HealthCountsStream healthStream;
            synchronized (HealthCountsStream.class) {
                HealthCountsStream existingStream = streams.get(commandKey.name());
                if (existingStream == null) {
                    // appendEventToBucket是一个Func2,负责将Hystrix各个事件转换成一个Bucket
                    HealthCountsStream newStream = new HealthCountsStream(commandKey, numBuckets, bucketSizeInMs,
                            HystrixCommandMetrics.appendEventToBucket);

                    streams.putIfAbsent(commandKey.name(), newStream);
                    healthStream = newStream;
                } else {
                    healthStream = existingStream;
                }
            }
            healthStream.startCachingStreamValuesIfUnstarted();
            return healthStream;
        }
    }

HealthCountsStream有两个父类,HealthCountsStream extends BucketedRollingCounterStream extends BucketedCounterStream,利用父类将stream的基础数据汇总成Bucket,再汇总成rolling window,最后得到统计结果HealthClounts.

下面按顺序看一下它和父类的调用情况:

首先是 BucketedCounterStream

public abstract class BucketedCounterStream<Event extends HystrixEvent, Bucket, Output> {
        ...
        protected BucketedCounterStream(final HystrixEventStream<Event> inputEventStream, final int numBuckets, final int bucketSizeInMs,
                                        final Func2<Bucket, Event, Bucket> appendRawEventToBucket) {
            this.numBuckets = numBuckets;
            // 将Hystrix事件汇总成Bucket的处理者, 是一个Func1
            this.reduceBucketToSummary = new Func1<Observable<Event>, Observable<Bucket>>() {
                // 传入Event类型的数据源,汇总成Bucket类型的数据
                @Override
                public Observable<Bucket> call(Observable<Event> eventBucket) {
                    ...
                }
            };
            ...
            this.bucketedStream = Observable.defer(new Func0<Observable<Bucket>>() {
                @Override
                public Observable<Bucket> call() {
                    // inputEventStream 就是一直提到的HystrixEventStream, 通过observe()来获取数据源
                    return inputEventStream
                            .observe()
                            // 利用窗口函数,收集一个Bucket时间内的数据
                            .window(bucketSizeInMs, TimeUnit.MILLISECONDS) 
                            // 将数据汇总成一个Bucket
                            .flatMap(reduceBucketToSummary)
                            .startWith(emptyEventCountsToStart);      
                }
            });
        }
    }

通过BucketedCounterStream,将数据汇总成了以Bucket为单位的stream. 然后,BucketedRollingCounterStream基于Bucket的stream,继续实现滑动窗口逻辑

protected BucketedRollingCounterStream(HystrixEventStream<Event> stream, final int numBuckets, int bucketSizeInMs,
                                           final Func2<Bucket, Event, Bucket> appendRawEventToBucket,
                                           final Func2<Output, Bucket, Output> reduceBucket) {
        super(stream, numBuckets, bucketSizeInMs, appendRawEventToBucket);
        // Bucket汇总处理者
        Func1<Observable<Bucket>, Observable<Output>> reduceWindowToSummary = new Func1<Observable<Bucket>, Observable<Output>>() {
            @Override
            public Observable<Output> call(Observable<Bucket> window) {
                return window.scan(getEmptyOutputValue(), reduceBucket).skip(numBuckets);
            }
        };
        // 基于父类BucketedCounterStream已经汇总的bucketedStream
        this.sourceStream = bucketedStream  
                // 将N个Bucket进行汇总
                .window(numBuckets, 1)      
                // 汇总成一个窗口
                .flatMap(reduceWindowToSummary)
                ...
                .share()                      
                .onBackpressureDrop();
    }

现在再回到熔断器的逻辑:

private Subscription subscribeToStream() {
        return metrics.getHealthCountsStream()
                .observe()
                .subscribe(new Subscriber<HealthCounts>() {
                    ...
                }

metrics.getHealthCountsStream()拿到的是一个已经汇总成以 “rollingWindow” 为单位的统计数据,observe() 实际拿到的是BucketedRollingCounterStream的sourceStream。如下:

public abstract class BucketedRollingCounterStream<...> {
        private Observable<Output> sourceStream;
        protected BucketedRollingCounterStream(...)
            // sourceStream 已经是rollingWindow级别的统计数据
            this.sourceStream = bucketedStream     
                    .window(numBuckets, 1)    
                    .flatMap(reduceWindowToSummary)...
        }

        @Override
        public Observable<Output> observe() {
            return sourceStream;
        }
    }

熔断器就是利用最终的统计结果HealthCounts来判断是否进行熔断。

熔断器状态变化

熔断器有三种状态,如下:

enum Status {
        CLOSED, OPEN, HALF_OPEN;
    }

在Command的执行过程中,会调用HystrixCircuitBreaker的方法来更新状态。下面是几个重要的方法:

命令执行时,判断熔断器是否打开

Spring Cloud 源码学习之 Hystrix 工作原理 中有介绍 Hystrix 如何实现其防护机制。

// 是否允许执行
    public boolean attemptExecution() {
        // 熔断器配置了强制打开, 不允许执行命令
        if (properties.circuitBreakerForceOpen().get()) {
            return false;
        }
        // 熔断器配置了强制关闭, 允许执行
        if (properties.circuitBreakerForceClosed().get()) {
            return true;
        }
        // AtomicLong circuitOpened, -1是表示熔断器未打开
        if (circuitOpened.get() == -1) {
            return true;
        } else {
            // 熔断后,会拒绝所有命令一段时间(默认5s), 称为sleepWindow
            if (isAfterSleepWindow()) {
                // 过了sleepWindow后,将熔断器设置为"HALF_OPEN",允许第一个请求过去
                if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
                    return true;
                } else {
                    return false;
                }
            } else {
                return false;
            }
        }
    }

当Command成功执行结束时,会调用HystrixCircuitBreaker.markSuccess()来标记执行成功.

public void markSuccess() {
        // 如果是HALF_OPEN状态,则关闭熔断器
        if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
            // 重新开始统计metrics,抛弃所有原先的metrics信息
            metrics.resetStream();
            Subscription previousSubscription = activeSubscription.get();
            if (previousSubscription != null) {
                previousSubscription.unsubscribe();
            }
            Subscription newSubscription = subscribeToStream();
            activeSubscription.set(newSubscription);
            // circuitOpened设置为-1表示关闭熔断器
            circuitOpened.set(-1L);
        }
    }

当Command执行失败时, 如果熔断器属于HALF_OPEN状态,也就是熔断器刚过sleepWindow时间,尝试放一个请求过去,结果又失败了,于是马上打开熔断器,继续拒绝sleepWindow的时间。

public void markNonSuccess() {
        if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {
            circuitOpened.set(System.currentTimeMillis());
        }
    }

这是调用markNonSuccess()的地方,handleFallback是所有失败情况的处理者.

final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
        @Override
        public Observable<R> call(Throwable t) {
            circuitBreaker.markNonSuccess();
            Exception e = getExceptionFromThrowable(t);
            executionResult = executionResult.setExecutionException(e);
            // 线程池拒绝
            if (e instanceof RejectedExecutionException) {
                return handleThreadPoolRejectionViaFallback(e);
            // 超时
            } else if (t instanceof HystrixTimeoutException) {
                return handleTimeoutViaFallback();
            // Bad Request 
            } else if (t instanceof HystrixBadRequestException) {
                return handleBadRequestByEmittingError(e);
            } else {
                if (e instanceof HystrixBadRequestException) {
                    eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
                    return Observable.error(e);
                }
                return handleFailureViaFallback(e);
            }
        }
    };

小结

断断续续写了好几天才写完,写作不易。Circuit-Breaker的设计、实现都很有意思:

  • 滴水成河,收集每个命令的执行情况,汇总后通过滑动窗口,不断动态计算最新统计数据,基于统计数据来开启熔断器
  • 巧妙的利用RxJava的window()函数来汇总数据,先汇总为Bucket, N Bucket组成Rolling Window
  • 使用sleepWindow + 尝试机制,自动恢复 “供电”

欢迎关注陈同学的公众号,一起学习,一起成长

20191123100240\_4.png


来源:http://ddrv.cn/a/88268

点赞(0)
版权归原创作者所有,任何形式转载请联系作者; Java 技术驿站 >> Spring Cloud 源码学习之 Hystrix 熔断器

相关推荐