Hystrix运行原理

1. 构造一个HystrixCommand或HystrixObservableCommand对象
2. 执行命令。
3. 检查是否已命中缓存，如果命中直接返回。
4. 检查断路器开关是否打开，如果打开，直接熔断，走fallback逻辑。
5. 检查线程池/队列/信号量是否已满，如果已满，直接拒绝请求，走fallback逻辑。
6. 上面条件都不满足，调用HystrixObservableCommand.construct()方法HystrixCommand.run()方法，执行业务逻辑。
7. 判断运行业务逻辑方法是否出现异常或者超时，如果出现，直接降级，走fallback逻辑。
8. 上报统计数据，用户计算断路器状态。
9. 返回结果

从流程图可以发现，只有出现5和7两种情况时，才会上报错误统计数据。

断路器运行原理

断路器的开关控制逻辑如下：

1. 在一个统计时间窗口内（HystrixCommandProperties.metricsRollingStatisticalWindowInMilliseconds()），处理的请求数量达到设定的最小阈值（HystrixCommandProperties.circuitBreakerRequestVolumeThreshold()），并且错误百分比超过设定的最大阈值（HystrixCommandProperties.circuitBreakerErrorThresholdPercentage()），这时断路器开关就会打开，断路器状态从转换CLOSED切换为OPEN。
2. 当断路器为打开状态时，它他会直接熔断所有请求（快速失败），走fallback逻辑。
3. 经过一个睡眠窗口时间后（HystrixCommandProperties.circuitBreakerSleepWindowInMilliseconds()），Hystrix会放行一个请到后续服务，并将断路器开关切换为半开状态（HALF-OPEN）。如果该请求失败，则断路器会将熔断开关切换为打开状态（OPEN），继续熔断所有请求，直到下一个睡眠时间窗口的到来；如果该请求成功，则断路器会切换到关闭状态（CLOSED）,这时将允许所有请求通过，直到出现1步骤的情况，断路器开关会切换为打开状态(OPEN)。

断路器源码

Hystrix断路器的实现类是HystrixCircuitBreaker，源码如下：

1/**
2 * Circuit-breaker logic that is hooked into {@link HystrixCommand} execution and will stop allowing executions if failures have gone past the defined threshold.
3 * 断路器，在HystrixCommand执行时会调用断路器逻辑，如果故障超过定义的阈值，断路器熔断开关将会打开，这时将阻止任务执行。
4 * <p>
5 * The default (and only) implementation  will then allow a single retry after a defined sleepWindow until the execution
6 * succeeds at which point it will again close the circuit and allow executions again.
7 * <p>
8 * 默认（且唯一）实现将允许在定义的sleepWindow之后进行单次重试，直到执行成功，此时它将再次关闭电路并允许再次执行。
9 */
10public interface HystrixCircuitBreaker {
11
12    /**
13     * Every {@link HystrixCommand} requests asks this if it is allowed to proceed or not.  It is idempotent and does
14     * not modify any internal state, and takes into account the half-open logic which allows some requests through
15     * after the circuit has been opened
16     * <p>
17     * 每个HystrixCommand请求都会询问是否允许继续（当断路器开关为OPEN和HALF_OPEN都时返回false，当断路器开关是CLOSE时或者到了下一个睡眠窗口时返回true）。
18     * 它是幂等的，不会修改任何内部状态，并考虑到半开逻辑，当一个睡眠窗口到来时他会放行一些请求到后续逻辑
19     *
20     * @return boolean whether a request should be permitted (是否应允许请求)
21     */
22    boolean allowRequest();
23
24    /**
25     * Whether the circuit is currently open (tripped).
26     * 判断熔断开关是否打开（如果是OPEN或HALF_OPEN时都返回true，如果为CLOSE时返回false，无副作用，是幂等方式）。
27     *
28     * @return boolean state of circuit breaker（返回断路器的状态）
29     */
30    boolean isOpen();
31
32    /**
33     * Invoked on successful executions from {@link HystrixCommand} as part of feedback mechanism when in a half-open state.
34     * <p>
35     * 断路器在处于半开状态时，作为反馈机制的一部分，从HystrixCommand成功执行时调用。
36     */
37    void markSuccess();
38
39    /**
40     * Invoked on unsuccessful executions from {@link HystrixCommand} as part of feedback mechanism when in a half-open state.
41     * 断路器当处于半开状态时，作为反馈机制的一部分，从HystrixCommand执行不成功的调用。
42     */
43    void markNonSuccess();
44
45    /**
46     * Invoked at start of command execution to attempt an execution.  This is non-idempotent - it may modify internal
47     * state.
48     * <p>
49     * 在命令执行开始时调用以尝试执行，主要所用时判断该请求是否可以执行。这是非幂等的 - 它可能会修改内部状态。
50     */
51    boolean attemptExecution();
52}
53

断路器的默认实现就是它的一个内部类：

1/**
2 * @ExcludeFromJavadoc
3 * @ThreadSafe
4 */
5class Factory {
6    // String is HystrixCommandKey.name() (we can't use HystrixCommandKey directly as we can't guarantee it implements hashcode/equals correctly)
7    // key是HystrixCommandKey.name()（我们不能直接使用HystrixCommandKey，因为我们无法保证它正确实现hashcode / equals）
8    private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();
9
10    /**
11     * 根据HystrixCommandKey获取HystrixCircuitBreaker
12     * Get the {@link HystrixCircuitBreaker} instance for a given {@link HystrixCommandKey}.
13     * <p>
14     * This is thread-safe and ensures only 1 {@link HystrixCircuitBreaker} per {@link HystrixCommandKey}.
15     *
16     * @param key        {@link HystrixCommandKey} of {@link HystrixCommand} instance requesting the {@link HystrixCircuitBreaker}
17     * @param group      Pass-thru to {@link HystrixCircuitBreaker}
18     * @param properties Pass-thru to {@link HystrixCircuitBreaker}
19     * @param metrics    Pass-thru to {@link HystrixCircuitBreaker}
20     * @return {@link HystrixCircuitBreaker} for {@link HystrixCommandKey}
21     */
22    public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
23        // this should find it for all but the first time
24        // 根据HystrixCommandKey获取断路器
25        HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());
26        if (previouslyCached != null) {
27            return previouslyCached;
28        }
29
30        // if we get here this is the first time so we need to initialize
31
32        // Create and add to the map ... use putIfAbsent to atomically handle the possible race-condition of
33        // 2 threads hitting this point at the same time and let ConcurrentHashMap provide us our thread-safety
34        // If 2 threads hit here only one will get added and the other will get a non-null response instead.
35        // 第一次没有获取到断路器，那么我们需要取初始化它
36        // 这里直接利用ConcurrentHashMap的putIfAbsent方法，它是原子操作，加入有两个线程执行到这里，将会只有一个线程将值放到容器中
37        // 让我们省掉了加锁的步骤
38        HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics));
39        if (cbForCommand == null) {
40            // this means the putIfAbsent step just created a new one so let's retrieve and return it
41            return circuitBreakersByCommand.get(key.name());
42        } else {
43            // this means a race occurred and while attempting to 'put' another one got there before
44            // and we instead retrieved it and will now return it
45            return cbForCommand;
46        }
47    }
48
49    /**
50     * 根据HystrixCommandKey获取HystrixCircuitBreaker，如果没有返回NULL
51     * Get the {@link HystrixCircuitBreaker} instance for a given {@link HystrixCommandKey} or null if none exists.
52     *
53     * @param key {@link HystrixCommandKey} of {@link HystrixCommand} instance requesting the {@link HystrixCircuitBreaker}
54     * @return {@link HystrixCircuitBreaker} for {@link HystrixCommandKey}
55     */
56    public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) {
57        return circuitBreakersByCommand.get(key.name());
58    }
59
60    /**
61     * Clears all circuit breakers. If new requests come in instances will be recreated.
62     * 清除所有断路器。如果有新的请求将会重新创建断路器放到容器。
63     */
64    /* package */
65    static void reset() {
66        circuitBreakersByCommand.clear();
67    }
68}
69
70
71/**
72 * 默认的断路器实现
73 * The default production implementation of {@link HystrixCircuitBreaker}.
74 *
75 * @ExcludeFromJavadoc
76 * @ThreadSafe
77 */
78/* package */class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {
79    private final HystrixCommandProperties properties;
80    private final HystrixCommandMetrics metrics;
81
82    enum Status {
83        // 断路器状态，关闭，打开，半开
84        CLOSED, OPEN, HALF_OPEN;
85    }
86
87    // 赋值操作不是线程安全的。若想不用锁来实现，可以用AtomicReference<V>这个类，实现对象引用的原子更新。
88    // AtomicReference 原子引用，保证Status原子性修改
89    private final AtomicReference<Status> status = new AtomicReference<Status>(Status.CLOSED);
90    // 记录断路器打开的时间点（时间戳）,如果这个时间大于0表示断路器处于打开状态或半开状态
91    private final AtomicLong circuitOpened = new AtomicLong(-1);
92    private final AtomicReference<Subscription> activeSubscription = new AtomicReference<Subscription>(null);
93
94    protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
95        this.properties = properties;
96        this.metrics = metrics;
97
98        //On a timer, this will set the circuit between OPEN/CLOSED as command executions occur
99        // 在定时器上，当命令执行发生时，这将在OPEN / CLOSED之间设置电路
100        Subscription s = subscribeToStream();
101        activeSubscription.set(s);
102    }
103
104    private Subscription subscribeToStream() {
105        /*
106         * This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream
107         * 此流将重新计算运行状况流中每个onNext上的OPEN / CLOSED状态
108         */
109        return metrics.getHealthCountsStream()
110                .observe()
111                .subscribe(new Subscriber<HealthCounts>() {
112                    @Override
113                    public void onCompleted() {
114
115                    }
116
117                    @Override
118                    public void onError(Throwable e) {
119
120                    }
121
122                    @Override
123                    public void onNext(HealthCounts hc) {
124                        // check if we are past the statisticalWindowVolumeThreshold
125                        // 检查一个时间窗口内的最小请求数
126                        if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
127                            // we are not past the minimum volume threshold for the stat window,
128                            // so no change to circuit status.
129                            // if it was CLOSED, it stays CLOSED
130                            // IF IT WAS HALF-OPEN, WE NEED TO WAIT FOR A SUCCESSFUL COMMAND EXECUTION
131                            // if it was open, we need to wait for sleep window to elapse
132                            // 我们没有超过统计窗口的最小音量阈值，所以我们不会去改变断路器状态，如果是closed状态，他将保持这个状态
133                            // 如果是半开状态，那么她需要等到一个成功的 Command执行
134                            // 如果是打开状态，那么它需要等到这个时间窗口过去
135                        } else {
136                            // 检查错误比例阀值
137                            if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
138                                //we are not past the minimum error threshold for the stat window,
139                                // so no change to circuit status.
140                                // if it was CLOSED, it stays CLOSED
141                                // if it was half-open, we need to wait for a successful command execution
142                                // if it was open, we need to wait for sleep window to elapse
143                            } else {
144                                // our failure rate is too high, we need to set the state to OPEN
145                                // 我们的失败率太高，我们需要将状态设置为OPEN
146                                if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
147                                    circuitOpened.set(System.currentTimeMillis());
148                                }
149                            }
150                        }
151                    }
152                });
153    }
154
155    @Override
156    public void markSuccess() {
157        // 断路器是处理半开并且HystrixCommand执行成功，将状态设置成关闭
158        if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
159            //This thread wins the race to close the circuit - it resets the stream to start it over from 0
160            //该线程赢得了关闭电路的竞争 - 它重置流以从0开始
161            metrics.resetStream();
162            Subscription previousSubscription = activeSubscription.get();
163            if (previousSubscription != null) {
164                previousSubscription.unsubscribe();
165            }
166            Subscription newSubscription = subscribeToStream();
167            activeSubscription.set(newSubscription);
168            circuitOpened.set(-1L);
169        }
170    }
171
172    @Override
173    public void markNonSuccess() {
174        // 断路器是处理半开并且HystrixCommand执行成功，将状态设置成打开
175        if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {
176            //This thread wins the race to re-open the circuit - it resets the start time for the sleep window
177            // 该线程赢得了重新打开电路的竞争 - 它重置了睡眠窗口的开始时间
178            circuitOpened.set(System.currentTimeMillis());
179        }
180    }
181
182    @Override
183    public boolean isOpen() {
184        // 获取配置判断断路器是否强制打开
185        if (properties.circuitBreakerForceOpen().get()) {
186            return true;
187        }
188        // 获取配置判断断路器是否强制关闭
189        if (properties.circuitBreakerForceClosed().get()) {
190            return false;
191        }
192        return circuitOpened.get() >= 0;
193    }
194
195    @Override
196    public boolean allowRequest() {
197        // 获取配置判断断路器是否强制打开
198        if (properties.circuitBreakerForceOpen().get()) {
199            return false;
200        }
201        // 获取配置判断断路器是否强制关闭
202        if (properties.circuitBreakerForceClosed().get()) {
203            return true;
204        }
205        if (circuitOpened.get() == -1) {
206            return true;
207        } else {
208            // 如果是半开状态则返回不允许Command执行
209            if (status.get().equals(Status.HALF_OPEN)) {
210                return false;
211            } else {
212                // 检查睡眠窗口是否过了
213                return isAfterSleepWindow();
214            }
215        }
216    }
217
218    private boolean isAfterSleepWindow() {
219        final long circuitOpenTime = circuitOpened.get();
220        final long currentTime = System.currentTimeMillis();
221        // 获取配置的一个睡眠的时间窗口
222        final long sleepWindowTime = properties.circuitBreakerSleepWindowInMilliseconds().get();
223        return currentTime > circuitOpenTime + sleepWindowTime;
224    }
225
226    @Override
227    public boolean attemptExecution() {
228        // 获取配置判断断路器是否强制打开
229        if (properties.circuitBreakerForceOpen().get()) {
230            return false;
231        }
232        // 获取配置判断断路器是否强制关闭
233        if (properties.circuitBreakerForceClosed().get()) {
234            return true;
235        }
236        if (circuitOpened.get() == -1) {
237            return true;
238        } else {
239            if (isAfterSleepWindow()) {
240                //only the first request after sleep window should execute
241                //if the executing command succeeds, the status will transition to CLOSED
242                //if the executing command fails, the status will transition to OPEN
243                //if the executing command gets unsubscribed, the status will transition to OPEN
244                // 只有一个睡眠窗口后的第一个请求会被执行
245                // 如果执行命令成功，状态将转换为CLOSED
246                // 如果执行命令失败，状态将转换为OPEN
247                // 如果执行命令取消订阅，状态将过渡到OPEN
248                if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
249                    return true;
250                } else {
251                    return false;
252                }
253            } else {
254                return false;
255            }
256        }
257    }
258}

• isOpen()：判断熔断开关是否打开（该方法是否幂等和Hystrix版本相关）。
• allowRequest()：每个HystrixCommand请求都会询问是否允许继续执行（当断路器开关为OPEN和HALF_OPEN都时返回false，当断路器开关是CLOSE或到了下一个睡眠窗口时返回true），它是幂等的，不会修改任何内部状态，并考虑到半开逻辑，当一个睡眠窗口到来时他会放行一些请求到后续逻辑。
• attemptExecution()：在命令执行开始时调用以尝试执行，主要所用时判断该请求是否可以执行。这是非幂等的，它可能会修改内部状态。

这里需要注意的是allowRequest()方法时幂等的，可以重复调用；attemptExecution()方法是有副作用的，不可以重复调用；isOpen()是否幂等和Hystrix版本有关。