Shulou(Shulou.com)06/02 Report--

This article will explain in detail how to analyze the core principles of Hystrix and the source code of circuit breakers. The content of the article is of high quality, so the editor will share it with you for reference. I hope you will have a certain understanding of the relevant knowledge after reading this article.

Operation principle of Hystrix

Construct a HystrixCommand or HystrixObservableCommand object

Carry out the order.

Check to see if the cache has been hit and return directly if it is hit.

Check whether the circuit breaker switch is turned on, if so, fuse directly and follow the fallback logic.

Check whether the thread pool / queue / semaphore is full, and if so, reject the request directly and follow the fallback logic.

If none of the above conditions are met, the HystrixObservableCommand.construct () method HystrixCommand.run () method is called to execute the business logic.

Determine whether there is an exception or timeout in running the business logic method, and if so, directly downgrade and follow the fallback logic.

Report statistics, and the user calculates the status of the circuit breaker.

Return the result

It can be found from the flowchart that error statistics are reported only when there are 5 and 7 cases.

Operation principle of circuit breaker

The switch control logic of the circuit breaker is as follows:

In a statistical time window (HystrixCommandProperties.metricsRollingStatisticalWindowInMilliseconds ()), when the number of requests processed reaches the set minimum threshold (HystrixCommandProperties.circuitBreakerRequestVolumeThreshold ()), and the percentage of errors exceeds the set maximum threshold (HystrixCommandProperties.circuitBreakerErrorThresholdPercentage ()), the circuit breaker switch is turned on and the circuit breaker state changes from transition CLOSED to OPEN.

When the circuit breaker is on, it will directly fuse all requests (quick failure) and follow the fallback logic.

After a sleep window time (HystrixCommandProperties.circuitBreakerSleepWindowInMilliseconds ()), Hystrix releases a request for follow-up service and switches the circuit breaker switch to half-open (HALF-OPEN). If the request fails, the circuit breaker switches the fuse switch to the on state (OPEN) and continues to fuse all requests until the next sleep time window arrives; if the request is successful, the circuit breaker switches to the off state (CLOSED), which allows all requests to pass until step 1 occurs, and the circuit breaker switch switches to the on state (OPEN).

Circuit breaker source code

The implementation class of Hystrix circuit breaker is HystrixCircuitBreaker, and the source code is as follows:

/ * * Circuit-breaker logic that is hooked into {@ link HystrixCommand} execution and will stop allowing executions if failures have gone past the defined threshold. * Circuit breaker, the circuit breaker logic will be invoked during HystrixCommand execution. If the fault exceeds the defined threshold, the circuit breaker fuse switch will be turned on, which will prevent the task from being executed. *

* The default (and only) implementation will then allow a single retry after a defined sleepWindow until the execution * succeeds at which point it will again close the circuit and allow executions again. *

* the default (and only) implementation will allow a single retry after the defined sleepWindow until the execution succeeds, at which point it will shut down the circuit again and allow it to execute again. * / public interface HystrixCircuitBreaker {/ * * Every {@ link HystrixCommand} requests asks this if it is allowed to proceed or not. It is idempotent and does * not modify any internal state, and takes into account the half-open logic which allows some requests through * after the circuit has been opened *

* each HystrixCommand request asks if it is allowed to continue (false is returned when the circuit breaker switch is both OPEN and HALF_OPEN, and true is returned when the circuit breaker switch is CLOSE or to the next sleep window). * it is idempotent and does not modify any internal state, and taking into account the half-open logic, when a sleep window arrives, it will release some requests to the subsequent logic * * @ return boolean whether a request should be permitted (whether the request should be allowed) * / boolean allowRequest (); / * * Whether the circuit is currently open (tripped). * determine whether the fuse switch is enabled (return true if it is OPEN or HALF_OPEN, and return false if it is CLOSE, which has no side effects and is idempotent). * * @ return boolean state of circuit breaker (returns the status of the circuit breaker) * / boolean isOpen (); / * * Invoked on successful executions from {@ link HystrixCommand} as part of feedback mechanism when in a half-open state. *

* when the circuit breaker is in a half-open state, it is called from the HystrixCommand when it is successfully executed as part of the feedback mechanism. * / void markSuccess (); / * * Invoked on unsuccessful executions from {@ link HystrixCommand} as part of feedback mechanism when in a half-open state. * when the circuit breaker is in a half-open state, it makes an unsuccessful call from HystrixCommand as part of the feedback mechanism. * / void markNonSuccess (); / * * Invoked at start of command execution to attempt an execution. This is non-idempotent-it may modify internal * state. *

* called at the beginning of command execution to try to execute, mainly to determine whether the request can be executed. This is non-idempotent-it may modify the internal state. * / boolean attemptExecution ();}

The default implementation of a circuit breaker is one of its inner classes:

/ * * @ ExcludeFromJavadoc * @ ThreadSafe * / class Factory {/ / String is HystrixCommandKey.name () (we can't use HystrixCommandKey directly as we can't guarantee it implements hashcode/equals correctly) / / key is HystrixCommandKey.name () (we cannot use HystrixCommandKey directly because we cannot guarantee that it implements hashcode/equals correctly) private static ConcurrentHashMap circuitBreakersByCommand = new ConcurrentHashMap () / * * obtain HystrixCircuitBreaker * Get the {@ link HystrixCircuitBreaker} instance for a given {@ link HystrixCommandKey} according to HystrixCommandKey. *

* This is thread-safe and ensures only 1 {@ link HystrixCircuitBreaker} per {@ link HystrixCommandKey}. * * @ param key {@ link HystrixCommandKey} of {@ link HystrixCommand} instance requesting the {@ link HystrixCircuitBreaker} * @ param group Pass-thru to {@ link HystrixCircuitBreaker} * @ param properties Pass-thru to {@ link HystrixCircuitBreaker} * @ param metrics Pass-thru to {@ link HystrixCircuitBreaker} * @ return {@ link HystrixCircuitBreaker} for {@ link HystrixCommandKey} / public static HystrixCircuitBreaker getInstance (HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties HystrixCommandMetrics metrics) {/ / this should find it for all but the first time / / get the circuit breaker according to HystrixCommandKey HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get (key.name ()) If (previouslyCached! = null) {return previouslyCached;} / / if we get here this is the first time so we need to initialize / / Create and add to the map... Use putIfAbsent to atomically handle the possible race-condition of / / 2 threads hitting this point at the same time and let ConcurrentHashMap provide us our thread-safety / / If 2 threads hit here only one will get added and the other will get a non-null response instead. / / if we didn't get the circuit breaker for the first time, we need to initialize it / / here we directly use the putIfAbsent method of ConcurrentHashMap, which is an atomic operation. If two threads are added to the container, only one thread will put the value into the container / / so that we can omit the locking step HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent (key.name (), new HystrixCircuitBreakerImpl (key, group, properties, metrics)). If (cbForCommand = = null) {/ / this means the putIfAbsent step just created a new one so let's retrieve and return it return circuitBreakersByCommand.get (key.name ());} else {/ / this means a race occurred and while attempting to 'put' another one got there before / / and we instead retrieved it and will now return it return cbForCommand }} / * get HystrixCircuitBreaker according to HystrixCommandKey. If NULL * Get the {@ link HystrixCircuitBreaker} instance for a given {@ link HystrixCommandKey} or null if none exists is not returned. * * @ param key {@ link HystrixCommandKey} of {@ link HystrixCommand} instance requesting the {@ link HystrixCircuitBreaker} * @ return {@ link HystrixCircuitBreaker} for {@ link HystrixCommandKey} * / public static HystrixCircuitBreaker getInstance (HystrixCommandKey key) {return circuitBreakersByCommand.get (key.name ());} / * Clears all circuit breakers. If new requests come in instances will be recreated. * clear all circuit breakers. If there is a new request, the circuit breaker will be recreated and placed in the container. * / * package * / static void reset () {circuitBreakersByCommand.clear ();}} / * default circuit breaker implementation * The default production implementation of {@ link HystrixCircuitBreaker}. * * @ ExcludeFromJavadoc * @ ThreadSafe * / / * package * / class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {private final HystrixCommandProperties properties; private final HystrixCommandMetrics metrics; enum Status {/ / Circuit Breaker status, off, open, half open CLOSED, OPEN, HALF_OPEN;} / / assignment operation is not thread safe. If you want to do this without locks, you can use the AtomicReference class to implement atomic updates to object references. / / AtomicReference atomic reference to ensure Status atomicity modification private final AtomicReference status = new AtomicReference (Status.CLOSED); / / record the time point when the circuit breaker is open (timestamp). If this time is greater than 0, it means that the circuit breaker is in the open state or half-open state private final AtomicLong circuitOpened = new AtomicLong (- 1); private final AtomicReference activeSubscription = new AtomicReference (null) 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 / / on the timer, when command execution occurs, this will set the circuit Subscription s = subscribeToStream (); activeSubscription.set (s) between OPEN/CLOSED } private Subscription subscribeToStream () {/ * * This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream * this flow will recalculate the OPEN/CLOSED status on each onNext in the health flow * / return metrics.getHealthCountsStream () .observe () .subscribe (new Subscriber () { @ Override public void onCompleted () {} @ Override public void onError (Throwable e) {} @ Override public void onNext (HealthCounts hc) {/ / check If we are past the statisticalWindowVolumeThreshold / / check the minimum number of requests in a time window if (hc.getTotalRequests ())

< properties.circuitBreakerRequestVolumeThreshold().get()) { // we are not past the minimum volume threshold for the stat window, // so no change to circuit status. // if it was CLOSED, it stays CLOSED // IF IT WAS HALF-OPEN, WE NEED TO WAIT FOR A SUCCESSFUL COMMAND EXECUTION // if it was open, we need to wait for sleep window to elapse // 我们没有超过统计窗口的最小音量阈值，所以我们不会去改变断路器状态，如果是closed状态，他将保持这个状态 // 如果是半开状态，那么她需要等到一个成功的 Command执行 // 如果是打开状态，那么它需要等到这个时间窗口过去 } else { // 检查错误比例阀值 if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) { //we are not past the minimum error threshold for the stat window, // so no change to circuit status. // if it was CLOSED, it stays CLOSED // if it was half-open, we need to wait for a successful command execution // if it was open, we need to wait for sleep window to elapse } else { // our failure rate is too high, we need to set the state to OPEN // 我们的失败率太高，我们需要将状态设置为OPEN if (status.compareAndSet(Status.CLOSED, Status.OPEN)) { circuitOpened.set(System.currentTimeMillis()); } } } } }); } @Override public void markSuccess() { // 断路器是处理半开并且HystrixCommand执行成功，将状态设置成关闭 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 //该线程赢得了关闭电路的竞争 - 它重置流以从0开始 metrics.resetStream(); Subscription previousSubscription = activeSubscription.get(); if (previousSubscription != null) { previousSubscription.unsubscribe(); } Subscription newSubscription = subscribeToStream(); activeSubscription.set(newSubscription); circuitOpened.set(-1L); } } @Override public void markNonSuccess() { // 断路器是处理半开并且HystrixCommand执行成功，将状态设置成打开 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()); } } @Override public boolean isOpen() { // 获取配置判断断路器是否强制打开 if (properties.circuitBreakerForceOpen().get()) { return true; } // 获取配置判断断路器是否强制关闭 if (properties.circuitBreakerForceClosed().get()) { return false; } return circuitOpened.get() >

= 0;} @ Override public boolean allowRequest () {/ / get the configuration to determine whether the circuit breaker forcibly turns on if (properties.circuitBreakerForceOpen (). Get ()) {return false;} / / get the configuration to determine whether the circuit breaker forcefully turns off if (properties.circuitBreakerForceClosed (). Get ()) {return true } if (circuitOpened.get () = =-1) {return true;} else {/ / if it is half open, Command is not allowed to execute if (status.get () .equals (Status.HALF_OPEN)) {return false } else {/ / check whether the sleep window has passed return isAfterSleepWindow ();} private boolean isAfterSleepWindow () {final long circuitOpenTime = circuitOpened.get (); final long currentTime = System.currentTimeMillis () / / get a configured sleep window final long sleepWindowTime = properties.circuitBreakerSleepWindowInMilliseconds () .get (); return currentTime > circuitOpenTime + sleepWindowTime;} @ Override public boolean attemptExecution () {/ / get the configuration to determine whether the circuit breaker forces if (properties.circuitBreakerForceOpen () .get ()) {return false } / / get the configuration to determine whether the circuit breaker forcibly shuts down if (properties.circuitBreakerForceClosed (). Get ()) {return true;} if (circuitOpened.get () =-1) {return true } else {if (isAfterSleepWindow ()) {/ / only the first request after sleep window should execute / / if the executing command succeeds, the status will transition to CLOSED / / if the executing command fails, the status will transition to OPEN / / if the executing command gets unsubscribed The status will transition to OPEN / / only the first request after a sleep window will be executed / / if the command is successful The state will be converted to CLOSED / / if the command fails, the state will be converted to OPEN / / if the command is executed to unsubscribe, the state will transition to OPEN if (status.compareAndSet (Status.OPEN, Status.HALF_OPEN)) {return true } else {return false;}} else {return false;}

IsOpen (): determines whether the fuse switch is turned on (whether the method is idempotent and related to the Hystrix version).

AllowRequest (): each HystrixCommand request asks if execution is allowed (return false when the circuit breaker switch is both OPEN and HALF_OPEN, return true when the circuit breaker switch is CLOSE or get to the next sleep window), it is idempotent, does not modify any internal state, and allows for half-open logic, it releases some requests to subsequent logic when a sleep window arrives.

AttemptExecution (): called at the beginning of command execution to try to execute, mainly to determine whether the request can be executed. This is non-idempotent and may modify the internal state.

It is important to note that the allowRequest () method is idempotent and can be called repeatedly; the attemptExecution () method has side effects and cannot be called repeatedly; and whether isOpen () is idempotent is related to the Hystrix version.

On how to analyze the core principles of Hystrix and circuit breaker source code to share here, I hope that the above content can be of some help to you, can learn more knowledge. If you think the article is good, you can share it for more people to see.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.