Spring Cloud 源码学习之 Hystrix 熔断器

文中源码基于 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 收集，这是上文的图。

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

原理

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

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

这是Hystrix官网的一张图：

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

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

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

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

相关配置

默认配置都在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属性。

1. abstract class AbstractCommand<R> implements HystrixInvokableInfo<R>, HystrixObservable<R> {

2.    protected final HystrixCircuitBreaker circuitBreaker;

3. }

在AbstractCommand构造器中初始化circuitBreaker。

1. private static HystrixCircuitBreaker initCircuitBreaker(boolean enabled, HystrixCircuitBreaker fromConstructor, HystrixCommandKey commandKey...) {

2.    // 如果启用了熔断器

3.    if (enabled) {

4.        // 若commandKey没有对应的CircuitBreaker,则创建

5.        if (fromConstructor == null) {

6.            return HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics);

7.        } else {

8.            // 如果有则返回现有的

9.            return fromConstructor;

10.        }

11.    } else {

12.        return new NoOpCircuitBreaker();

13.    }

14. }

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

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

1. public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {

2.    // 如果有则返回现有的, key.name()即command的name作为检索条件

3.    HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());

4.    if (previouslyCached != null) {

5.        return previouslyCached;

6.    }

7.    // 如果没有则创建并cache

8.    HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics));

9.    if (cbForCommand == null) {

10.        return circuitBreakersByCommand.get(key.name());

11.    } else {

12.        return cbForCommand;

13.    }

14. }

如何订阅HystrixEventStream

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

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

1. protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {

2.    this.properties = properties;

3.    // 这是Command中的metrics对象,metrics对象也是commandKey维度的

4.    this.metrics = metrics;

5.    // !!!重点:订阅事件流

6.    Subscription s = subscribeToStream();

7.    activeSubscription.set(s);

8. }

9. // 订阅事件流, 前面打的比方: 小溪汇成的河, 各事件以结构化数据汇入了Stream中

10. private Subscription subscribeToStream() {

11.    // HealthCountsStream是重点，下面会分析

12.    return metrics.getHealthCountsStream()

13.            .observe()

14.            // 利用数据统计的结果HealthCounts, 实现熔断器

15.            .subscribe(new Subscriber<HealthCounts>() {

16.                @Override

17.                public void onCompleted() {}

18.                @Override

19.                public void onError(Throwable e) {}

20.                @Override

21.                public void onNext(HealthCounts hc) {

22.                    // 检查是否达到最小请求数,默认20个; 未达到的话即使请求全部失败也不会熔断

23.                    if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {

24.                        // 啥也不做

25.                    } else {

26.                        // 错误百分比未达到设定的阀值

27.                        if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {

28.                        } else {

29.                            // 错误率过高, 进行熔断

30.                            if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {

31.                                circuitOpened.set(System.currentTimeMillis());

32.                            }

33.                        }

34.                    }

35.                }

36.            });

37. }

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

1. public static class HealthCounts {

2.    // rolling window 中请求总数量

3.    private final long totalCount;

4.    // 错误请求数(failure + success + timeout + threadPoolRejected + semaphoreRejected)

5.    private final long errorCount;

6.    // 错误率

7.    private final int errorPercentage;

8. }

滑动窗口的实现

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

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

1. private HealthCountsStream healthCountsStream;

2. HystrixCommandMetrics(final HystrixCommandKey key, ...) {

3.    // 又是以key为维度

4.    healthCountsStream = HealthCountsStream.getInstance(key, properties);

5. }

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

1. Maintains a stream of rolling health counts for a given Command.

2. 它是一份流数据, 承载了指定Command的健康统计数据

3. There is a rolling window abstraction on this stream.

4. 基于这个stream, 抽象出了滑动窗口

5. The HealthCounts object is calculated over a window of t1 milliseconds.  This window has b buckets.

6. HealthCounts的数据是根据一个t1毫秒的滑动窗口计算得来,这个窗口有b个Buckets

7. Therefore, a new HealthCounts object is produced every t2 (=t1/b) milliseconds

8. 因此, 每 t2 (=t1/b)毫秒就会产生一个HealthCounts作为统计结果

HealthCountsStream.getInstance如下：

1. public static HealthCountsStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties) {

2.    // 每个Bucket的时间长度

3.    final int healthCountBucketSizeInMs = properties.metricsHealthSnapshotIntervalInMilliseconds().get();

4.     // 滑动窗口的时间/每个Bucket的时间长度=滑动窗口内Bucket的数量

5.    final int numHealthCountBuckets = properties.metricsRollingStatisticalWindowInMilliseconds().get() / healthCountBucketSizeInMs;

6.    return getInstance(commandKey, numHealthCountBuckets, healthCountBucketSizeInMs);

7. }

8. // 以Key为维度,每个Key有自己唯一的一个HealthCountsStream

9. public static HealthCountsStream getInstance(HystrixCommandKey commandKey, int numBuckets, int bucketSizeInMs) {

10.    HealthCountsStream initialStream = streams.get(commandKey.name());

11.    if (initialStream != null) {

12.        return initialStream;

13.    } else {

14.        final HealthCountsStream healthStream;

15.        synchronized (HealthCountsStream.class) {

16.            HealthCountsStream existingStream = streams.get(commandKey.name());

17.            if (existingStream == null) {

18.                // appendEventToBucket是一个Func2,负责将Hystrix各个事件转换成一个Bucket

19.                HealthCountsStream newStream = new HealthCountsStream(commandKey, numBuckets, bucketSizeInMs,

20.                        HystrixCommandMetrics.appendEventToBucket);

21.                streams.putIfAbsent(commandKey.name(), newStream);

22.                healthStream = newStream;

23.            } else {

24.                healthStream = existingStream;

25.            }

26.        }

27.        healthStream.startCachingStreamValuesIfUnstarted();

28.        return healthStream;

29.    }

30. }

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

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

首先是 BucketedCounterStream：

1. public abstract class BucketedCounterStream<Event extends HystrixEvent, Bucket, Output> {

2.    ...

3.    protected BucketedCounterStream(final HystrixEventStream<Event> inputEventStream, final int numBuckets, final int bucketSizeInMs,

4.                                    final Func2<Bucket, Event, Bucket> appendRawEventToBucket) {

5.        this.numBuckets = numBuckets;

6.        // 将Hystrix事件汇总成Bucket的处理者, 是一个Func1

7.        this.reduceBucketToSummary = new Func1<Observable<Event>, Observable<Bucket>>() {

8.            // 传入Event类型的数据源,汇总成Bucket类型的数据

9.            @Override

10.            public Observable<Bucket> call(Observable<Event> eventBucket) {

11.                ...

12.            }

13.        };

14.        ...

15.        this.bucketedStream = Observable.defer(new Func0<Observable<Bucket>>() {

16.            @Override

17.            public Observable<Bucket> call() {

18.                // inputEventStream 就是一直提到的HystrixEventStream, 通过observe()来获取数据源

19.                return inputEventStream

20.                        .observe()

21.                        // 利用窗口函数,收集一个Bucket时间内的数据

22.                        .window(bucketSizeInMs, TimeUnit.MILLISECONDS)

23.                        // 将数据汇总成一个Bucket

24.                        .flatMap(reduceBucketToSummary)

25.                        .startWith(emptyEventCountsToStart);      

26.            }

27.        });

28.    }

29. }

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

1. protected BucketedRollingCounterStream(HystrixEventStream<Event> stream, final int numBuckets, int bucketSizeInMs,

2.                                       final Func2<Bucket, Event, Bucket> appendRawEventToBucket,

3.                                       final Func2<Output, Bucket, Output> reduceBucket) {

4.    super(stream, numBuckets, bucketSizeInMs, appendRawEventToBucket);

5.    // Bucket汇总处理者

6.    Func1<Observable<Bucket>, Observable<Output>> reduceWindowToSummary = new Func1<Observable<Bucket>, Observable<Output>>() {

7.        @Override

8.        public Observable<Output> call(Observable<Bucket> window) {

9.            return window.scan(getEmptyOutputValue(), reduceBucket).skip(numBuckets);

10.        }

11.    };

12.    // 基于父类BucketedCounterStream已经汇总的bucketedStream

13.    this.sourceStream = bucketedStream  

14.            // 将N个Bucket进行汇总

15.            .window(numBuckets, 1)      

16.            // 汇总成一个窗口

17.            .flatMap(reduceWindowToSummary)

18.            ...

19.            .share()                      

20.            .onBackpressureDrop();

21. }

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

1. private Subscription subscribeToStream() {

2.    return metrics.getHealthCountsStream()

3.            .observe()

4.            .subscribe(new Subscriber<HealthCounts>() {

5.                ...

6.            }

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

1. public abstract class BucketedRollingCounterStream<...> {

2.    private Observable<Output> sourceStream;

3.    protected BucketedRollingCounterStream(...)

4.        // sourceStream 已经是rollingWindow级别的统计数据

5.        this.sourceStream = bucketedStream    

6.                .window(numBuckets, 1)    

7.                .flatMap(reduceWindowToSummary)...

8.    }

9.    @Override

10.    public Observable<Output> observe() {

11.        return sourceStream;

12.    }

13. }

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

熔断器状态变化

熔断器有三种状态，如下：

1. enum Status {

2.    CLOSED, OPEN, HALF_OPEN;

3. }

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

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

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

1. // 是否允许执行

2. public boolean attemptExecution() {

3.    // 熔断器配置了强制打开, 不允许执行命令

4.    if (properties.circuitBreakerForceOpen().get()) {

5.        return false;

6.    }

7.    // 熔断器配置了强制关闭, 允许执行

8.    if (properties.circuitBreakerForceClosed().get()) {

9.        return true;

10.    }

11.    // AtomicLong circuitOpened, -1是表示熔断器未打开

12.    if (circuitOpened.get() == -1) {

13.        return true;

14.    } else {

15.        // 熔断后,会拒绝所有命令一段时间(默认5s), 称为sleepWindow

16.        if (isAfterSleepWindow()) {

17.            // 过了sleepWindow后,将熔断器设置为"HALF_OPEN",允许第一个请求过去

18.            if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {

19.                return true;

20.            } else {

21.                return false;

22.            }

23.        } else {

24.            return false;

25.        }

26.    }

27. }

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

1. public void markSuccess() {

2.    // 如果是HALF_OPEN状态,则关闭熔断器

3.    if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {

4.        // 重新开始统计metrics,抛弃所有原先的metrics信息

5.        metrics.resetStream();

6.        Subscription previousSubscription = activeSubscription.get();

7.        if (previousSubscription != null) {

8.            previousSubscription.unsubscribe();

9.        }

10.        Subscription newSubscription = subscribeToStream();

11.        activeSubscription.set(newSubscription);

12.        // circuitOpened设置为-1表示关闭熔断器

13.        circuitOpened.set(-1L);

14.    }

15. }

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

1. public void markNonSuccess() {

2.    if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {

3.        circuitOpened.set(System.currentTimeMillis());

4.    }

5. }

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

1. final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {

2.    @Override

3.    public Observable<R> call(Throwable t) {

4.        circuitBreaker.markNonSuccess();

5.        Exception e = getExceptionFromThrowable(t);

6.        executionResult = executionResult.setExecutionException(e);

7.        // 线程池拒绝

8.        if (e instanceof RejectedExecutionException) {

9.            return handleThreadPoolRejectionViaFallback(e);

10.        // 超时

11.        } else if (t instanceof HystrixTimeoutException) {

12.            return handleTimeoutViaFallback();

13.        // Bad Request    

14.        } else if (t instanceof HystrixBadRequestException) {

15.            return handleBadRequestByEmittingError(e);

16.        } else {

17.            if (e instanceof HystrixBadRequestException) {

18.                eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);

19.                return Observable.error(e);

20.            }

21.            return handleFailureViaFallback(e);

22.        }

23.    }

24. };

小结

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

• 滴水成河，收集每个命令的执行情况，汇总后通过滑动窗口，不断动态计算最新统计数据，基于统计数据来开启熔断器

• 巧妙的利用RxJava的window()函数来汇总数据，先汇总为Bucket， N Bucket组成Rolling Window

• 使用sleepWindow + 尝试机制，自动恢复 “供电”