Hystrix 熔断器执行和超时实现机制

Posted 程序员泥瓦匠

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Hystrix 熔断器执行和超时实现机制相关的知识，希望对你有一定的参考价值。

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

1.Hystrix 熔断器类结构

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

2.Hystrix 熔断器状态

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

3.代码视角

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

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

3.1 如何判定开启熔断
 
   
   
 
    
  
    
    
  
  1. protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
    2. 
  
    
    
  
  2.     this.properties = properties;
    3. 
  
    
    
  
  3.     this.metrics = metrics;
    4. 
  
    
    
  

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

    10. 
  
    
    
  
  10. private Subscription subscribeToStream() {
    11. 
  
    
    
  
  11.     /*
    12. 
  
    
    
  
  12.      * This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream
    13. 
  
    
    
  
  13.      */
    14. 
  
    
    
  
  14.     return metrics.getHealthCountsStream()
    15. 
  
    
    
  
  15.             .observe()
    16. 
  
    
    
  
  16.             .subscribe(new Subscriber<HealthCounts>() {
    17. 
  
    
    
  

    18. 
  
    
    
  
  18.                  //.....................省略干扰代码......................
    19. 
  
    
    
  
  19.                 @Override
    20. 
  
    
    
  
  20.                 public void onNext(HealthCounts hc) {
    21. 
  
    
    
  
  21.                     // check if we are past the statisticalWindowVolumeThreshold
    22. 
  
    
    
  
  22.                     if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
    23. 
  
    
    
  

    24. 
  
    
    
  
  24.                     } else {
    25. 
  
    
    
  
  25.                         if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
    26. 
  
    
    
  

    27. 
  
    
    
  
  27.                         } else {
    28. 
  
    
    
  

    29. 
  
    
    
  
  29.                             if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
    30. 
  
    
    
  
  30.                                 circuitOpened.set(System.currentTimeMillis());
    31. 
  
    
    
  
  31.                             }
    32. 
  
    
    
  
  32.                         }
    33. 
  
    
    
  
  33.                     }
    34. 
  
    
    
  
  34.                 }
    35. 
  
    
    
  
  35.             });
    36. 
  
    
    
  
  36. }
    37.

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

3.2 attemptExecution

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

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

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

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

 
   
   
 
    
  
    
    
  
  1. public boolean attemptExecution() {
    2. 
  
    
    
  
  2.             if (properties.circuitBreakerForceOpen().get()) {
    3. 
  
    
    
  
  3.                 return false;
    4. 
  
    
    
  
  4.             }
    5. 
  
    
    
  
  5.             if (properties.circuitBreakerForceClosed().get()) {
    6. 
  
    
    
  
  6.                 return true;
    7. 
  
    
    
  
  7.             }
    8. 
  
    
    
  
  8.             if (circuitOpened.get() == -1) {
    9. 
  
    
    
  
  9.                 return true;
    10. 
  
    
    
  
  10.             } else {
    11. 
  
    
    
  
  11.                 if (isAfterSleepWindow()) {
    12. 
  
    
    
  
  12.                     if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
    13. 
  
    
    
  
  13.                         //only the first request after sleep window should execute
    14. 
  
    
    
  
  14.                         return true;
    15. 
  
    
    
  
  15.                     } else {
    16. 
  
    
    
  
  16.                         return false;
    17. 
  
    
    
  
  17.                     }
    18. 
  
    
    
  
  18.                 } else {
    19. 
  
    
    
  
  19.                     return false;
    20. 
  
    
    
  
  20.                 }
    21. 
  
    
    
  
  21.             }
    22. 
  
    
    
  
  22.         }
    23.

这里代码判断逻辑 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

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

    4. 
  
    
    
  
  4.     final Action1<R> markEmits = new Action1<R>() {
    5. 
  
    
    
  
  5.         @Override
    6. 
  
    
    
  
  6.         public void call(R r) {
    7. 
  
    
    
  
  7.             if (shouldOutputOnNextEvents()) {
    8. 
  
    
    
  
  8.                 executionResult = executionResult.addEvent(HystrixEventType.EMIT);
    9. 
  
    
    
  
  9.                 eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
    10. 
  
    
    
  
  10.             }
    11. 
  
    
    
  
  11.             if (commandIsScalar()) {
    12. 
  
    
    
  
  12.                 ......省略干扰代码.......
    13. 
  
    
    
  
  13.                 circuitBreaker.markSuccess();
    14. 
  
    
    
  
  14.             }
    15. 
  
    
    
  
  15.         }
    16. 
  
    
    
  
  16.     };
    17. 
  
    
    
  

    18. 
  
    
    
  
  18.     final Action0 markOnCompleted = new Action0() {
    19. 
  
    
    
  
  19.         @Override
    20. 
  
    
    
  
  20.         public void call() {
    21. 
  
    
    
  
  21.             if (!commandIsScalar()) {
    22. 
  
    
    
  
  22.                 ......省略干扰代码.......
    23. 
  
    
    
  
  23.                 circuitBreaker.markSuccess();
    24. 
  
    
    
  
  24.             }
    25. 
  
    
    
  
  25.         }
    26. 
  
    
    
  
  26.     };
    27. 
  
    
    
  

    28. 
  
    
    
  
  28.     final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
    29. 
  
    
    
  
  29.         @Override
    30. 
  
    
    
  
  30.         public Observable<R> call(Throwable t) {
    31. 
  
    
    
  
  31.             circuitBreaker.markNonSuccess();
    32. 
  
    
    
  
  32.             ......省略干扰代码.......
    33. 
  
    
    
  
  33.         }
    34. 
  
    
    
  
  34.     };
    35. 
  
    
    
  

    36. 
  
    
    
  
  36.     ......省略干扰代码.......
    37. 
  
    
    
  

    38. 
  
    
    
  
  38.     return execution.doOnNext(markEmits)
    39. 
  
    
    
  
  39.             .doOnCompleted(markOnCompleted)
    40. 
  
    
    
  
  40.             .onErrorResumeNext(handleFallback)
    41. 
  
    
    
  
  41.             .doOnEach(setRequestContext);
    42. 
  
    
    
  
  42. }
    43.

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

markNonSuccess

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

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

markSuccess

 
   
   
 
    
  
    
    
  
  1. @Override
    2. 
  
    
    
  
  2. public void markSuccess() {
    3. 
  
    
    
  
  3.     if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
    4. 
  
    
    
  
  4.         //This thread wins the race to close the circuit - it resets the stream to start it over from 0
    5. 
  
    
    
  
  5.         metrics.resetStream();
    6. 
  
    
    
  
  6.         Subscription previousSubscription = activeSubscription.get();
    7. 
  
    
    
  
  7.         if (previousSubscription != null) {
    8. 
  
    
    
  
  8.             previousSubscription.unsubscribe();
    9. 
  
    
    
  
  9.         }
    10. 
  
    
    
  
  10.         Subscription newSubscription = subscribeToStream();
    11. 
  
    
    
  
  11.         activeSubscription.set(newSubscription);
    12. 
  
    
    
  
  12.         circuitOpened.set(-1L);
    13. 
  
    
    
  
  13.     }
    14. 
  
    
    
  
  14. }
    15.

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

4.THE END

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

HystrixCommand在执行的过程中如何探测超时，本篇主要对此进行介绍说明。

1.主入口：executeCommandAndObserve

 
   
   
 
    
  
    
    
  
  1. #com.netflix.hystrix.AbstractCommand#executeCommandAndObserve
    2. 
  
    
    
  
  2. private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
    3. 
  
    
    
  
  3.         ···省略部分代码···
    4. 
  
    
    
  
  4.         Observable<R> execution;
    5. 
  
    
    
  

    6. 
  
    
    
  
  6.         //判断是否开启超时监测
    7. 
  
    
    
  
  7.         if (properties.executionTimeoutEnabled().get()) {
    8. 
  
    
    
  
  8.             execution = executeCommandWithSpecifiedIsolation(_cmd)
    9. 
  
    
    
  
  9.                     .lift(new HystrixObservableTimeoutOperator<R>(_cmd));
    10. 
  
    
    
  
  10.         } else {
    11. 
  
    
    
  
  11.             execution = executeCommandWithSpecifiedIsolation(_cmd);
    12. 
  
    
    
  
  12.         }
    13. 
  
    
    
  

    14. 
  
    
    
  
  14.         return execution.doOnNext(markEmits)
    15. 
  
    
    
  
  15.                 .doOnCompleted(markOnCompleted)
    16. 
  
    
    
  
  16.                 .onErrorResumeNext(handleFallback)
    17. 
  
    
    
  
  17.                 .doOnEach(setRequestContext);
    18. 
  
    
    
  
  18.     }
    19.

executeCommandWithSpecifiedIsolation(cmd) .lift(new HystrixObservableTimeoutOperator ( cmd));

可以简单的认为lift 里面的对前面的Observable包含，类似装饰者，后面的parent就是指上层的Observable。其中 HystrixObservableTimeoutOperator 就是关键的部分。

2.关键点: HystrixObservableTimeoutOperator

先看下HystrixObservableTimeoutOperator.call()，TimerListener的实现

 
   
   
 
    
  
    
    
  
  1. TimerListener listener = new TimerListener() {
    2. 
  
    
    
  

    3. 
  
    
    
  
  3.                 @Override
    4. 
  
    
    
  
  4.                 public void tick() {
    5. 
  
    
    
  

    6. 
  
    
    
  
  6.                     if (originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, TimedOutStatus.TIMED_OUT)) {
    7. 
  
    
    
  
  7.                         // 标记事件，可以认为是开的hook，这里暂忽略
    8. 
  
    
    
  
  8.                         originalCommand.eventNotifier.markEvent(HystrixEventType.TIMEOUT, originalCommand.commandKey);
    9. 
  
    
    
  

    10. 
  
    
    
  
  10.                         //取消原Obserable的订阅
    11. 
  
    
    
  
  11.                         s.unsubscribe();
    12. 
  
    
    
  

    13. 
  
    
    
  
  13.                         final HystrixContextRunnable timeoutRunnable = new HystrixContextRunnable(originalCommand.concurrencyStrategy, hystrixRequestContext, new Runnable() {
    14. 
  
    
    
  

    15. 
  
    
    
  
  15.                             @Override
    16. 
  
    
    
  
  16.                             public void run() {
    17. 
  
    
    
  
  17.                                 child.onError(new HystrixTimeoutException());
    18. 
  
    
    
  
  18.                             }
    19. 
  
    
    
  
  19.                         });
    20. 
  
    
    
  
  20.                         timeoutRunnable.run();
    21. 
  
    
    
  
  21.                     }
    22. 
  
    
    
  
  22.                 }
    23. 
  
    
    
  

    24. 
  
    
    
  
  24.                 //获取配置的超时时间配置
    25. 
  
    
    
  
  25.                 @Override
    26. 
  
    
    
  
  26.                 public int getIntervalTimeInMilliseconds() {
    27. 
  
    
    
  
  27.                     return originalCommand.properties.executionTimeoutInMilliseconds().get();
    28. 
  
    
    
  
  28.                 }
    29. 
  
    
    
  
  29.             };
    30.

这段代码的意思就是，给当前command的超时状态置为超时，如果设置成功就抛出 HystrixTimeoutException异常，紧接着被command的 doOnErron接收走 fallback逻辑

 
   
   
 
    
  
    
    
  
  1. fallback
    2. 
  
    
    
  
  2. private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
    3. 
  
    
    
  
  3.         final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
    4. 
  
    
    
  

    5. 
  
    
    
  
  5.         .................................
    6. 
  
    
    
  

    7. 
  
    
    
  
  7.         final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
    8. 
  
    
    
  
  8.             @Override
    9. 
  
    
    
  
  9.             public Observable<R> call(Throwable t) {
    10. 
  
    
    
  
  10.                 circuitBreaker.markNonSuccess();
    11. 
  
    
    
  
  11.                 Exception e = getExceptionFromThrowable(t);
    12. 
  
    
    
  
  12.                 executionResult = executionResult.setExecutionException(e);
    13. 
  
    
    
  
  13.                 if (e instanceof RejectedExecutionException) {
    14. 
  
    
    
  
  14.                     return handleThreadPoolRejectionViaFallback(e);
    15. 
  
    
    
  
  15.                 } else if (t instanceof HystrixTimeoutException) {
    16. 
  
    
    
  
  16.                     //此处catch到超时异常
    17. 
  
    
    
  
  17.                     return handleTimeoutViaFallback();
    18. 
  
    
    
  
  18.                 } else if (t instanceof HystrixBadRequestException) {
    19. 
  
    
    
  
  19.                     return handleBadRequestByEmittingError(e);
    20. 
  
    
    
  
  20.                 } else {
    21. 
  
    
    
  
  21.                     /*
    22. 
  
    
    
  
  22.                      * Treat HystrixBadRequestException from ExecutionHook like a plain HystrixBadRequestException.
    23. 
  
    
    
  
  23.                      */
    24. 
  
    
    
  
  24.                     if (e instanceof HystrixBadRequestException) {
    25. 
  
    
    
  
  25.                         eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
    26. 
  
    
    
  
  26.                         return Observable.error(e);
    27. 
  
    
    
  
  27.                     }
    28. 
  
    
    
  

    29. 
  
    
    
  
  29.                     return handleFailureViaFallback(e);
    30. 
  
    
    
  
  30.                 }
    31. 
  
    
    
  
  31.             }
    32. 
  
    
    
  
  32.         };
    33. 
  
    
    
  

    34. 
  
    
    
  
  34.         .................................
    35. 
  
    
    
  

    36. 
  
    
    
  
  36.         return execution.doOnNext(markEmits)
    37. 
  
    
    
  
  37.                 .doOnCompleted(markOnCompleted)
    38. 
  
    
    
  
  38.                 .onErrorResumeNext(handleFallback)
    39. 
  
    
    
  
  39.                 .doOnEach(setRequestContext);
    40. 
  
    
    
  
  40.     }
    41.

同时 s.unsubscribe()通知正在执行的线程，终止任务。如何终止呢？

executeCommandWithSpecifiedIsolation.subscribeOn()

subscribeOne的参数就是 HystrixContextScheduler, Rxjava里 scheduler具体干活的是 worker，我们先看下Hystrix自定义scheduler的结构示意图 Hystrix 熔断器执行和超时实现机制 那么我们直奔主题，直接看 ThreadPoolWorker

 
   
   
 
    
  
    
    
  
  1. //ThreadPoolWorker.schedule
    2. 
  
    
    
  
  2. @Override
    3. 
  
    
    
  
  3. public Subscription schedule(final Action0 action) {
    4. 
  
    
    
  
  4.     if (subscription.isUnsubscribed()) {
    5. 
  
    
    
  
  5.         return Subscriptions.unsubscribed();
    6. 
  
    
    
  
  6.     }
    7. 
  
    
    
  

    8. 
  
    
    
  
  8.     ScheduledAction sa = new ScheduledAction(action);
    9. 
  
    
    
  

    10. 
  
    
    
  
  10.     subscription.add(sa);
    11. 
  
    
    
  
  11.     sa.addParent(subscription);
    12. 
  
    
    
  

    13. 
  
    
    
  
  13.     ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();
    14. 
  
    
    
  
  14.     FutureTask<?> f = (FutureTask<?>) executor.submit(sa);
    15. 
  
    
    
  
  15.     sa.add(new FutureCompleterWithConfigurableInterrupt(f, shouldInterruptThread, executor));
    16. 
  
    
    
  

    17. 
  
    
    
  
  17.     return sa;
    18. 
  
    
    
  
  18. }
    19.

1.开始的时候判断observable是否被订阅 2.被订阅后，将任务 submit到线程池 3. FutureCompleterWithConfigurableInterrupt scheduler在执行的时候，增加了observable的中断探测

 
   
   
 
    
  
    
    
  
  1. private static class FutureCompleterWithConfigurableInterrupt implements Subscription {
    2. 
  
    
    
  
  2.     private final FutureTask<?> f;
    3. 
  
    
    
  
  3.     private final Func0<Boolean> shouldInterruptThread;
    4. 
  
    
    
  
  4.     private final ThreadPoolExecutor executor;
    5. 
  
    
    
  

    6. 
  
    
    
  
  6.     private FutureCompleterWithConfigurableInterrupt(FutureTask<?> f, Func0<Boolean> shouldInterruptThread, ThreadPoolExecutor executor) {
    7. 
  
    
    
  
  7.         this.f = f;
    8. 
  
    
    
  
  8.         this.shouldInterruptThread = shouldInterruptThread;
    9. 
  
    
    
  
  9.         this.executor = executor;
    10. 
  
    
    
  
  10.     }
    11. 
  
    
    
  

    12. 
  
    
    
  
  12.     @Override
    13. 
  
    
    
  
  13.     public void unsubscribe() {
    14. 
  
    
    
  
  14.         executor.remove(f);
    15. 
  
    
    
  
  15.         if (shouldInterruptThread.call()) {
    16. 
  
    
    
  
  16.             f.cancel(true);
    17. 
  
    
    
  
  17.         } else {
    18. 
  
    
    
  
  18.             f.cancel(false);
    19. 
  
    
    
  
  19.         }
    20. 
  
    
    
  
  20.     }
    21. 
  
    
    
  

    22. 
  
    
    
  
  22.     .....省略代码.......
    23. 
  
    
    
  
  23. }
    24.

当observable 取消订阅时，就会把当前任务移除，并中断任务

到这里只是讲说了超时后的处理，如何认定执行超时呢？

3.匠心之巧

这里有个很巧妙的设计，再探 HystrixObservableTimeoutOperator

 
   
   
 
    
  
    
    
  
  1. final Reference<TimerListener> tl = HystrixTimer.getInstance().addTimerListener(listener);
    2. 
  
    
    
  

    3. 
  
    
    
  
  3. #com.netflix.hystrix.util.HystrixTimer#addTimerListener
    4. 
  
    
    
  
  4. public Reference<TimerListener> addTimerListener(final TimerListener listener) {
    5. 
  
    
    
  
  5.         startThreadIfNeeded();
    6. 
  
    
    
  
  6.         // add the listener
    7. 
  
    
    
  

    8. 
  
    
    
  
  8.         Runnable r = new Runnable() {
    9. 
  
    
    
  

    10. 
  
    
    
  
  10.             @Override
    11. 
  
    
    
  
  11.             public void run() {
    12. 
  
    
    
  
  12.                 try {
    13. 
  
    
    
  
  13.                     listener.tick();
    14. 
  
    
    
  
  14.                 } catch (Exception e) {
    15. 
  
    
    
  
  15.                     logger.error("Failed while ticking TimerListener", e);
    16. 
  
    
    
  
  16.                 }
    17. 
  
    
    
  
  17.             }
    18. 
  
    
    
  
  18.         };
    19. 
  
    
    
  

    20. 
  
    
    
  
  20.         ScheduledFuture<?> f = executor.get().getThreadPool().scheduleAtFixedRate(r, listener.getIntervalTimeInMilliseconds(), listener.getIntervalTimeInMilliseconds(), TimeUnit.MILLISECONDS);
    21. 
  
    
    
  
  21.         return new TimerReference(listener, f);
    22. 
  
    
    
  
  22.     }
    23.

利用了 ScheduledThreadPoolExecutor，延迟执行，延迟时间就是我们设定的超时时间，我们再看下

 
   
   
 
    
  
    
    
  
  1. #HystrixObservableTimeoutOperator
    2. 
  
    
    
  
  2. Subscriber<R> parent = new Subscriber<R>() {
    3. 
  
    
    
  

    4. 
  
    
    
  
  4.                 @Override
    5. 
  
    
    
  
  5.                 public void onCompleted() {
    6. 
  
    
    
  
  6.                     if (isNotTimedOut()) {
    7. 
  
    
    
  
  7.                         // stop timer and pass notification through
    8. 
  
    
    
  
  8.                         tl.clear();
    9. 
  
    
    
  
  9.                         child.onCompleted();
    10. 
  
    
    
  
  10.                     }
    11. 
  
    
    
  
  11.                 }
    12. 
  
    
    
  

    13. 
  
    
    
  
  13.                 @Override
    14. 
  
    
    
  
  14.                 public void onError(Throwable e) {
    15. 
  
    
    
  
  15.                     if (isNotTimedOut()) {
    16. 
  
    
    
  
  16.                         // stop timer and pass notification through
    17. 
  
    
    
  
  17.                         tl.clear();
    18. 
  
    
    
  
  18.                         child.onError(e);
    19. 
  
    
    
  
  19.                     }
    20. 
  
    
    
  
  20.                 }
    21. 
  
    
    
  

    22. 
  
    
    
  
  22.                 .....  .....  .....  .....  .....  .....  .....  .....  .....
    23. 
  
    
    
  

    24. 
  
    
    
  
  24.                 private boolean isNotTimedOut() {
    25. 
  
    
    
  
  25.                     // if already marked COMPLETED (by onNext) or succeeds in setting to COMPLETED
    26. 
  
    
    
  
  26.                     return originalCommand.isCommandTimedOut.get() == TimedOutStatus.COMPLETED ||
    27. 
  
    
    
  
  27.                             originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, TimedOutStatus.COMPLETED);
    28. 
  
    
    
  
  28.                 }
    29. 
  
    
    
  

    30. 
  
    
    
  
  30.             };
    31.

这里parent就是指上层的obserable，这里可以抽象的认为是我们的HystrixCommand执行线程, 当command执行线程执行完成的时候或异常的时候，会执行 tl.clear()， 也就是Future.cancel()会中断 TimerListener 的ScheduledFuture 线程，迫使超时机制失效。

 
   
   
 
    
  
    
    
  
  1. // tl.clear()
    2. 
  
    
    
  
  2. private static class TimerReference extends SoftReference<TimerListener> {
    3. 
  
    
    
  
  3.         private final ScheduledFuture<?> f;
    4. 
  
    
    
  
  4.         ....        ....        ....        ....        ....
    5. 
  
    
    
  
  5.         @Override
    6. 
  
    
    
  
  6.         public void clear() {
    7. 
  
    
    
  
  7.             super.clear();
    8. 
  
    
    
  
  8.             // stop this ScheduledFuture from any further executions
    9. 
  
    
    
  
  9.             f.cancel(false);
    10. 
  
    
    
  
  10.         }
    11. 
  
    
    
  
  11.     }
    12.

4.回归文字

HystrixCommand里有个 TimedOutStatus 超时状态 Hystrix 熔断器执行和超时实现机制 现在可以认为有两个线程，一个是hystrixCommand任务执行线程，一个是等着给hystrixCommand判定超时的线程，现在两个线程看谁能先把hystrixCommand的状态置换，只要任何一个线程对hystrixCommand打上标就意味着超时判定结束。

相关热门推荐文章：



长按二维码，扫扫关注哦

✬如果你喜欢这篇文章，欢迎分享和点赞✬

以上是关于Hystrix 熔断器执行和超时实现机制的主要内容，如果未能解决你的问题，请参考以下文章

Hystrix熔断机制导致误报请求超时错误

深入浅出SpringCloud原理及实战「Netflix系列之Hystrix」针对于限流熔断组件Hystrix的回退降级实现方案和机制

深入浅出SpringCloud原理及实战「Netflix系列之Hystrix」针对于限流熔断组件Hystrix的请求合并机制实现原理分析

降级特技之使用Hystrix实现降级和熔断—《亿级流量网站架构核心技术》

springCloud-Hystrix熔断器

深入浅出SpringCloud原理及实战「Netflix系列之Hystrix」针对于限流熔断组件Hystrix的原理和实现机制