Hystrix熔断器执行机制

2018-03-22 本文已影响175人青芒v5

本篇假设大家对Hystrix的执行过程及源码有一定的了解，这里介绍Hystrix的熔断器执行机制。

1.Hystrix 熔断器类结构

HystrixCircuitBreaker作为接口定义，具体的实现有NoOpCircuitBreaker和HystrixCircuitBreakerImpl，其中NoOpCircuitBreaker只是个空壳没有具体的实现，相当于不熔断。HystrixCircuitBreakerImpl是主要的熔断逻辑实现。

2.Hystrix 熔断器状态

熔断器有三个状态 CLOSED、OPEN、HALF_OPEN 熔断器默认关闭状态，当触发熔断后状态变更为 OPEN,在等待到指定的时间，Hystrix会放请求检测服务是否开启，这期间熔断器会变为HALF_OPEN 半开启状态，熔断探测服务可用则继续变更为 CLOSED关闭熔断器。

状态变更示意图

3.代码视角

ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();

Hystrix为每个commandKey都维护了一个熔断器，保持着对应的熔断器，所以当new XXXHystrixCommand()的时候依然能够保持着原来熔断器的状态。

3.1 如何判定开启熔断

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

    //On a timer, this will set the circuit between OPEN/CLOSED as command executions occur
    Subscription s = subscribeToStream();
    activeSubscription.set(s);
}

private Subscription subscribeToStream() {
    /*
     * This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream
     */
    return metrics.getHealthCountsStream()
            .observe()
            .subscribe(new Subscriber<HealthCounts>() {
                
                 //.....................省略干扰代码......................
                @Override
                public void onNext(HealthCounts hc) {
                    // check if we are past the statisticalWindowVolumeThreshold
                    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());
                            }
                        }
                    }
                }
            });
}

这里面HystrixBreaker启动的时候会订阅HystrixCommandMetrics的 HealthCountsStream，每当HealthCountsStream搜集到数据，都会触发上面的 onNext方法，然后该方法做下面几个判断
1.当前请求量是否达到设定水位（请求量太小不做阀值控制）
2.当前的请求错误量是否达到阀值，达到后会将熔断器状态置为 OPEN, circuitOpened设置为当前时间戳表示开启的时间。

3.2 attemptExecution

先看下HystrixCommand 的执行Observable
com.netflix.hystrix.AbstractCommand#applyHystrixSemantics

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
        // mark that we're starting execution on the ExecutionHook
        // if this hook throws an exception, then a fast-fail occurs with no fallback.  No state is left inconsistent
        executionHook.onStart(_cmd);

        /* determine if we're allowed to execute */
        if (circuitBreaker.attemptExecution()) {
··········省略代码··········

这里，每次HystrixCommand执行都会调用 circuitBreaker.attemptExecution()

public boolean attemptExecution() {
            if (properties.circuitBreakerForceOpen().get()) {
                return false;
            }
            if (properties.circuitBreakerForceClosed().get()) {
                return true;
            }
            if (circuitOpened.get() == -1) {
                return true;
            } else {
                if (isAfterSleepWindow()) {
                    if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
                        //only the first request after sleep window should execute
                        return true;
                    } else {
                        return false;
                    }
                } else {
                    return false;
                }
            }
        }

这里代码判断逻辑
1.判断是否强制开启熔断器，是则return false，command不能执行
2.判断是否强制关闭熔断器，是则return true, command可执行
3.判断熔断器是否开启 circuitOpened.get() == -1表示没有开启，则return true，command可执行。
4.到这步证明已经开启了熔断器，那么判断是否可尝试请求，如果可以同时会把熔断器的状态改为HALF_OPEN

3.3 markSuccess&markNonSuccess

com.netflix.hystrix.AbstractCommand#executeCommandAndObserve

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
    ......省略干扰代码.......

    final Action1<R> markEmits = new Action1<R>() {
        @Override
        public void call(R r) {
            if (shouldOutputOnNextEvents()) {
                executionResult = executionResult.addEvent(HystrixEventType.EMIT);
                eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
            }
            if (commandIsScalar()) {
                ......省略干扰代码.......
                circuitBreaker.markSuccess();
            }
        }
    };

    final Action0 markOnCompleted = new Action0() {
        @Override
        public void call() {
            if (!commandIsScalar()) {
                ......省略干扰代码.......
                circuitBreaker.markSuccess();
            }
        }
    };

    final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
        @Override
        public Observable<R> call(Throwable t) {
            circuitBreaker.markNonSuccess();
            ......省略干扰代码.......
        }
    };

    ......省略干扰代码.......

    return execution.doOnNext(markEmits)
            .doOnCompleted(markOnCompleted)
            .onErrorResumeNext(handleFallback)
            .doOnEach(setRequestContext);
}

此处表示HystrixCommand执行的过程中对应的熔断器状态变更，上面代码不难看出，当error的时候会触发circuitBreaker.markNonSuccess();，执行成功或者执行完成触发 circuitBreaker.markSuccess();

markNonSuccess

@Override
public void markNonSuccess() {
    if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {
        //This thread wins the race to re-open the circuit - it resets the start time for the sleep window
        circuitOpened.set(System.currentTimeMillis());
    }
}

如果能执行到markNonSuccess，说明此时熔断器是关闭状态，或者尝试放流阶段。关闭状态的话不做处理（未触发熔断），尝试放流时，发现依然执行失败，这里讲熔断器状态重新置为开启状态，并把circuitOpened设置为当前的时间戳。

markSuccess

@Override
public void markSuccess() {
    if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
        //This thread wins the race to close the circuit - it resets the stream to start it over from 0
        metrics.resetStream();
        Subscription previousSubscription = activeSubscription.get();
        if (previousSubscription != null) {
            previousSubscription.unsubscribe();
        }
        Subscription newSubscription = subscribeToStream();
        activeSubscription.set(newSubscription);
        circuitOpened.set(-1L);
    }
}

能走到markSuccess说明熔断器此时关闭或者放流阶段，尝试放流阶段则讲熔断器关闭，设置circuitOpened=-1，并重置指标统计。

4.THE END

到这里熔断器的介绍就结束了，回顾下主要有熔断器如何开启、如何关闭、几个状态的变更。一个完整的熔断器就此呈现在大家的面前。