Fork me on GitHub

framework/SpringCloud/Hystrix/hystrix使用

字数统计: 2.4k字   |   阅读时长≈ 13分 
1
2
3
4
5
6
7
8
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.netflix.hystrix.HystrixAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.ReactiveHystrixCircuitBreakerAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.security.HystrixSecurityAutoConfiguration

org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker=\
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerConfiguration

Hystrix

它有什么作用？

1）延迟和容错

停止级联故障。后备和正常降级。无法快速快速恢复。

使用断路器隔离线程和信号量。

2）实时操作

实时监控和配置更改。监视服务和财产变更在整个机队中扩散后立即生效。

在几秒钟内收到警报，做出决定，影响更改并查看结果。

3）并发

并行执行。并发请求缓存。通过请求折叠自动进行批处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

public class CommandHelloWorld extends HystrixCommand<String> {

    private final String name;

    public CommandHelloWorld(String name) {
        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
        this.name = name;
    }

    @Override
    protected String run() {
        return "Hello " + name + "!";
    }
}
This command could be used like this:

String s = new CommandHelloWorld("Bob").execute();
Future<String> s = new CommandHelloWorld("Bob").queue();
Observable<String> s = new CommandHelloWorld("Bob").observe();



可以看到，所有的执行都不是直接调用run方法，而是调用execute，queue，observe，所以不会直接执行，代码也不会阻塞，他会交给别的线程去执行

核心逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

public Observable<R> toObservable() {
      final AbstractCommand<R> _cmd = this;

      //doOnCompleted handler already did all of the SUCCESS work
      //doOnError handler already did all of the FAILURE/TIMEOUT/REJECTION/BAD_REQUEST work
    
    //设置一个终止命令
      final Action0 terminateCommandCleanup = new Action0() {

          @Override
          public void call() {
              if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.TERMINAL)) {
                  handleCommandEnd(false); //user code never ran
              } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.TERMINAL)) {
                  handleCommandEnd(true); //user code did run
              }
          }
      };

      //mark the command as CANCELLED and store the latency (in addition to standard cleanup)
    //设置一个取消订阅的命令
      final Action0 unsubscribeCommandCleanup = new Action0() {
          @Override
          public void call() {
              if (_cmd.commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.UNSUBSCRIBED)) {
                  if (!_cmd.executionResult.containsTerminalEvent()) {
                      _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
                      try {
                          executionHook.onUnsubscribe(_cmd);
                      } catch (Throwable hookEx) {
                          logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
                      }
                      _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
                              .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
                  }
                  handleCommandEnd(false); //user code never ran
              } else if (_cmd.commandState.compareAndSet(CommandState.USER_CODE_EXECUTED, CommandState.UNSUBSCRIBED)) {
                  if (!_cmd.executionResult.containsTerminalEvent()) {
                      _cmd.eventNotifier.markEvent(HystrixEventType.CANCELLED, _cmd.commandKey);
                      try {
                          executionHook.onUnsubscribe(_cmd);
                      } catch (Throwable hookEx) {
                          logger.warn("Error calling HystrixCommandExecutionHook.onUnsubscribe", hookEx);
                      }
                      _cmd.executionResultAtTimeOfCancellation = _cmd.executionResult
                              .addEvent((int) (System.currentTimeMillis() - _cmd.commandStartTimestamp), HystrixEventType.CANCELLED);
                  }
                  handleCommandEnd(true); //user code did run
              }
          }
      };

    
    //应用Hystrix语义
      final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
          @Override
          public Observable<R> call() {
              if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
                  return Observable.never();
              }
              return applyHystrixSemantics(_cmd);
          }
      };

    //包装所有的 钩子
      final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() {
          @Override
          public R call(R r) {
              R afterFirstApplication = r;

              try {
                  afterFirstApplication = executionHook.onComplete(_cmd, r);
              } catch (Throwable hookEx) {
                  logger.warn("Error calling HystrixCommandExecutionHook.onComplete", hookEx);
              }

              try {
                  return executionHook.onEmit(_cmd, afterFirstApplication);
              } catch (Throwable hookEx) {
                  logger.warn("Error calling HystrixCommandExecutionHook.onEmit", hookEx);
                  return afterFirstApplication;
              }
          }
      };

    //完成时发射的钩子
      final Action0 fireOnCompletedHook = new Action0() {
          @Override
          public void call() {
              try {
                  executionHook.onSuccess(_cmd);
              } catch (Throwable hookEx) {
                  logger.warn("Error calling HystrixCommandExecutionHook.onSuccess", hookEx);
              }
          }
      };

    //返回一个可观察调用可观察到的工厂来创建可观察到的每一个新的观察员，订阅。 也就是说，用于每个用户，实际可观察​​的是订户观察由工厂函数来确定。
      return Observable.defer(new Func0<Observable<R>>() {
          
          //推迟调用，就是通过返回的Observable来交给 订阅者处理
          @Override
          public Observable<R> call() {
               /* 这是一个有状态的对象，因此只能使用一次 */
              if (!commandState.compareAndSet(CommandState.NOT_STARTED, CommandState.OBSERVABLE_CHAIN_CREATED)) {
                  IllegalStateException ex = new IllegalStateException("This instance can only be executed once. Please instantiate a new instance.");
                  //TODO make a new error type for this
                  throw new HystrixRuntimeException(FailureType.BAD_REQUEST_EXCEPTION, _cmd.getClass(), getLogMessagePrefix() + " command executed multiple times - this is not permitted.", ex, null);
              }

              
              commandStartTimestamp = System.currentTimeMillis();

              if (properties.requestLogEnabled().get()) {
                  // log this command execution regardless of what happened
                  if (currentRequestLog != null) {
                      currentRequestLog.addExecutedCommand(_cmd);
                  }
              }

              final boolean requestCacheEnabled = isRequestCachingEnabled();
              final String cacheKey = getCacheKey();

              /* try from cache first */
              if (requestCacheEnabled) {
                  HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
                  if (fromCache != null) {
                      isResponseFromCache = true;
                      return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                  }
              }
              ---上面缓存获取
              

                 // --推迟应用 applyHystrixSemantics
              Observable<R> hystrixObservable =
                      Observable.defer(applyHystrixSemantics)
                              .map(wrapWithAllOnNextHooks);

              Observable<R> afterCache;

              // put in cache
              if (requestCacheEnabled && cacheKey != null) {
                  // wrap it for caching
                  HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
                  HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.putIfAbsent(cacheKey, toCache);
                  if (fromCache != null) {
                      // another thread beat us so we'll use the cached value instead
                      toCache.unsubscribe();
                      isResponseFromCache = true;
                      return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
                  } else {
                      // we just created an ObservableCommand so we cast and return it
                      afterCache = toCache.toObservable();
                  }
              } else {
                  afterCache = hystrixObservable;
              }

              
              //最后使用缓存进行最 终止，取消订阅和 完成设置 回调钩子
              return afterCache
                      .doOnTerminate(terminateCommandCleanup)     // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line))
                      .doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once
                      .doOnCompleted(fireOnCompletedHook);
          }
      });
  }

applyHystrixSemantics

断路器作用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

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
    
    //执行钩子回调onStart方法
        executionHook.onStart(_cmd);

        /* determine if we're allowed to execute */
    
    //查看断路器是否允许执行
        if (circuitBreaker.allowRequest()) {
            
            
            final TryableSemaphore executionSemaphore = getExecutionSemaphore();
            final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
            final Action0 singleSemaphoreRelease = new Action0() {
                @Override
                public void call() {
                    if (semaphoreHasBeenReleased.compareAndSet(false, true)) {
                        executionSemaphore.release();
                    }
                }
            };

            final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {
                @Override
                public void call(Throwable t) {
                    eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey);
                }
            };

            if (executionSemaphore.tryAcquire()) {
                try {
                    /* used to track userThreadExecutionTime */
                    executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
                    return executeCommandAndObserve(_cmd)
                            .doOnError(markExceptionThrown)
                            .doOnTerminate(singleSemaphoreRelease)
                            .doOnUnsubscribe(singleSemaphoreRelease);
                } catch (RuntimeException e) {
                    return Observable.error(e);
                }
            } else {
                return handleSemaphoreRejectionViaFallback();
            }
        } else {
            return handleShortCircuitViaFallback();
        }
    }

executeCommandAndObserve

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
       final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();

    
    //设置  发射的标记
       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()) {
                   long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
                   eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
                   eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
                   executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
                   circuitBreaker.markSuccess();
               }
           }
       };

    //设置 完成时的标记处理
       final Action0 markOnCompleted = new Action0() {
           @Override
           public void call() {
               if (!commandIsScalar()) {
                   long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
                   eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
                   eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
                   executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
                   circuitBreaker.markSuccess();
               }
           }
       };

    
    //设置 后备逻辑处理
       final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
           @Override
           public Observable<R> call(Throwable t) {
               Exception e = getExceptionFromThrowable(t);
               executionResult = executionResult.setExecutionException(e);
               if (e instanceof RejectedExecutionException) {
                   return handleThreadPoolRejectionViaFallback(e);
               } else if (t instanceof HystrixTimeoutException) {
                   return handleTimeoutViaFallback();
               } else if (t instanceof HystrixBadRequestException) {
                   return handleBadRequestByEmittingError(e);
               } else {
                   /*
                    * Treat HystrixBadRequestException from ExecutionHook like a plain HystrixBadRequestException.
                    */
                   if (e instanceof HystrixBadRequestException) {
                       eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
                       return Observable.error(e);
                   }

                   return handleFailureViaFallback(e);
               }
           }
       };

    
    //设置 请求的上下文处理
       final Action1<Notification<? super R>> setRequestContext = new Action1<Notification<? super R>>() {
           @Override
           public void call(Notification<? super R> rNotification) {
               setRequestContextIfNeeded(currentRequestContext);
           }
       };

    //是否开启超时
       Observable<R> execution;
       if (properties.executionTimeoutEnabled().get()) {
           
           //超时处理
           execution = executeCommandWithSpecifiedIsolation(_cmd)
                   .lift(new HystrixObservableTimeoutOperator<R>(_cmd));
       } else {
           execution = executeCommandWithSpecifiedIsolation(_cmd);
       }

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

executeCommandWithSpecifiedIsolation

隔离

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
      if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
          // mark that we are executing in a thread (even if we end up being rejected we still were a THREAD execution and not SEMAPHORE)
          //标记我们正在线程中执行（即使我们最终被拒绝，我们仍然是线程执行而不是信号量）
          return Observable.defer(new Func0<Observable<R>>() {
              @Override
              public Observable<R> call() {
                  executionResult = executionResult.setExecutionOccurred();
                  if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
                      return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
                  }

                  metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD);

                  if (isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT) {
                      // the command timed out in the wrapping thread so we will return immediately
                      // and not increment any of the counters below or other such logic
                      return Observable.error(new RuntimeException("timed out before executing run()"));
                  }
                  if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) {
                      //we have not been unsubscribed, so should proceed
                      HystrixCounters.incrementGlobalConcurrentThreads();
                      threadPool.markThreadExecution();
                      // store the command that is being run
                      endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
                      executionResult = executionResult.setExecutedInThread();
                      /**
                       * If any of these hooks throw an exception, then it appears as if the actual execution threw an error
                       */
                      try {
                          executionHook.onThreadStart(_cmd);
                          executionHook.onRunStart(_cmd);
                          executionHook.onExecutionStart(_cmd);
                          return getUserExecutionObservable(_cmd);
                      } catch (Throwable ex) {
                          return Observable.error(ex);
                      }
                  } else {
                      //command has already been unsubscribed, so return immediately
                      return Observable.error(new RuntimeException("unsubscribed before executing run()"));
                  }
              }
          }).doOnTerminate(new Action0() {
              @Override
              public void call() {
                  if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.TERMINAL)) {
                      handleThreadEnd(_cmd);
                  }
                  if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.TERMINAL)) {
                      //if it was never started and received terminal, then no need to clean up (I don't think this is possible)
                  }
                  //if it was unsubscribed, then other cleanup handled it
              }
          }).doOnUnsubscribe(new Action0() {
              @Override
              public void call() {
                  if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.UNSUBSCRIBED)) {
                      handleThreadEnd(_cmd);
                  }
                  if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.UNSUBSCRIBED)) {
                      //if it was never started and was cancelled, then no need to clean up
                  }
                  //if it was terminal, then other cleanup handled it
              }
          }).subscribeOn(threadPool.getScheduler(new Func0<Boolean>() {
              @Override
              public Boolean call() {
                  return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;
              }
          }));
      } else {
          return Observable.defer(new Func0<Observable<R>>() {
              @Override
              public Observable<R> call() {
                  executionResult = executionResult.setExecutionOccurred();
                  if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
                      return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
                  }

                  metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.SEMAPHORE);
                  //信号量隔离存储正在运行的命令
                  endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
                  try {
                      executionHook.onRunStart(_cmd);
                      executionHook.onExecutionStart(_cmd);
                      return getUserExecutionObservable(_cmd);  //getUserExecutionObservable方法已经包装了同步异常，因此不应引发
                  } catch (Throwable ex) {
                      //If the above hooks throw, then use that as the result of the run method
                      return Observable.error(ex);
                  }
              }
          });
      }
  }

getUserExecutionObservable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17


private Observable<R> getUserExecutionObservable(final AbstractCommand<R> _cmd) {
    Observable<R> userObservable;

    try {
        userObservable = getExecutionObservable();
    } catch (Throwable ex) {
        // the run() method is a user provided implementation so can throw instead of using Observable.onError
        // so we catch it here and turn it into Observable.error
        userObservable = Observable.error(ex);
    }

    return userObservable
            .lift(new ExecutionHookApplication(_cmd))
            .lift(new DeprecatedOnRunHookApplication(_cmd));
}

getExecutionObservable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Override
   final protected Observable<R> getExecutionObservable() {
       return Observable.defer(new Func0<Observable<R>>() {
           @Override
           public Observable<R> call() {
               try {
                   //最终 这里 是一个初始的被观察者，一层一层包着 装饰
                   return Observable.just(run());
               } catch (Throwable ex) {
                   return Observable.error(ex);
               }
           }
       }).doOnSubscribe(new Action0() {
           @Override
           public void call() {
               // Save thread on which we get subscribed so that we can interrupt it later if needed
               executionThread.set(Thread.currentThread());
           }
       });
   }
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家