SpringCloud Hystrix 源码解析 —— 命令执行（一）之正常执行逻辑

// `AbstractCommand#toObservable()` 方法
 1: final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
 2: @Override
 3: public Observable<R> call(){
 4: // commandState 处于 UNSUBSCRIBED 时，不执行命令
 5: if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
 6: return Observable.never();
 7: }
 8: // 获得 执行Observable
 9: return applyHystrixSemantics(_cmd);
 10: }
 11: };

 1: private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd){
 2: // TODO 【2003】【HOOK】
 3: // mark that we're starting execution on the ExecutionHook
 4: // if this hook throws an exception, then a fast-fail occurs with no fallback. No state is left inconsistent
 5: executionHook.onStart(_cmd);
 6: 
 7: /* determine if we're allowed to execute */
 8: if (circuitBreaker.attemptExecution()) {
 9: // 获得 信号量
10: final TryableSemaphore executionSemaphore = getExecutionSemaphore();
11: 
12: // 信号量释放Action
13: final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
14: final Action0 singleSemaphoreRelease = new Action0() {
15: @Override
16: public void call(){
17: if (semaphoreHasBeenReleased.compareAndSet(false, true)) {
18: executionSemaphore.release();
19: }
20: }
21: };
22: 
23: // TODO 【2011】【Hystrix 事件机制】
24: final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {
25: @Override
26: public void call(Throwable t){
27: eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey);
28: }
29: };
30: 
31: // 信号量 获得
32: if (executionSemaphore.tryAcquire()) {
33: try {
34: // 标记 executionResult 调用开始时间
35: /* used to track userThreadExecutionTime */
36: executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());
37: 
38: // 获得 执行Observable
39: return executeCommandAndObserve(_cmd)
40: .doOnError(markExceptionThrown)
41: .doOnTerminate(singleSemaphoreRelease)
42: .doOnUnsubscribe(singleSemaphoreRelease);
43: } catch (RuntimeException e) {
44: return Observable.error(e);
45: }
46: } else {
47: return handleSemaphoreRejectionViaFallback();
48: }
49: } else {
50: return handleShortCircuitViaFallback();
51: }
52: }

interface TryableSemaphore{
 
 boolean tryAcquire();
 
 void release();
 
 int getNumberOfPermitsUsed();
}

/* package */static class TryableSemaphoreNoOp implements TryableSemaphore{

 public static final TryableSemaphore DEFAULT = new TryableSemaphoreNoOp();

 @Override
 public boolean tryAcquire(){
 return true;
 }

 @Override
 public void release(){
 
 }

 @Override
 public int getNumberOfPermitsUsed(){
 return 0;
 }

}

/* package */static class TryableSemaphoreActual implements TryableSemaphore{
 protected final HystrixProperty<Integer> numberOfPermits;
 private final AtomicInteger count = new AtomicInteger(0);

 public TryableSemaphoreActual(HystrixProperty<Integer> numberOfPermits){
 this.numberOfPermits = numberOfPermits;
 }

 @Override
 public boolean tryAcquire(){
 int currentCount = count.incrementAndGet();
 if (currentCount > numberOfPermits.get()) {
 count.decrementAndGet();
 return false;
 } else {
 return true;
 }
 }

 @Override
 public void release(){
 count.decrementAndGet();
 }

 @Override
 public int getNumberOfPermitsUsed(){
 return count.get();
 }

}

/**
* 执行命令（正常执行）信号量映射
* KEY ：命令名 {@link #commandKey}
*/
/* each circuit has a semaphore to restrict concurrent fallback execution */
protected static final ConcurrentHashMap<String, TryableSemaphore> executionSemaphorePerCircuit = new ConcurrentHashMap<String, TryableSemaphore>();
 
protected TryableSemaphore getExecutionSemaphore(){
 if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.SEMAPHORE) {
 if (executionSemaphoreOverride == null) {
 TryableSemaphore _s = executionSemaphorePerCircuit.get(commandKey.name());
 if (_s == null) { // 不存在时，创建 TryableSemaphoreActual
 // we didn't find one cache so setup
 executionSemaphorePerCircuit.putIfAbsent(commandKey.name(), new TryableSemaphoreActual(properties.executionIsolationSemaphoreMaxConcurrentRequests()));
 // assign whatever got set (this or another thread)
 return executionSemaphorePerCircuit.get(commandKey.name());
 } else {
 return _s;
 }
 } else {
 return executionSemaphoreOverride;
 }
 } else {
 // return NoOp implementation since we're not using SEMAPHORE isolation
 return TryableSemaphoreNoOp.DEFAULT;
 }
}

 1: private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd){
 2: // TODO 【】
 3: final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
 4: 
 5: // TODO 【2007】【executionResult】用途
 6: final Action1<R> markEmits = new Action1<R>() {
 7: @Override
 8: public void call(R r){
 9: if (shouldOutputOnNextEvents()) {
10: executionResult = executionResult.addEvent(HystrixEventType.EMIT);
11: eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
12: }
13: if (commandIsScalar()) {
14: long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
15: eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
16: executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
17: eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
18: circuitBreaker.markSuccess();
19: }
20: }
21: };
22: 
23: // TODO 【2007】【executionResult】用途
24: final Action0 markOnCompleted = new Action0() {
25: @Override
26: public void call(){
27: if (!commandIsScalar()) {
28: long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
29: eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
30: executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
31: eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
32: circuitBreaker.markSuccess();
33: }
34: }
35: };
36: 
37: // 失败回退逻辑 Func1
38: final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
39: @Override
40: public Observable<R> call(Throwable t){
41: circuitBreaker.markNonSuccess();
42: Exception e = getExceptionFromThrowable(t);
43: executionResult = executionResult.setExecutionException(e);
44: if (e instanceof RejectedExecutionException) {
45: return handleThreadPoolRejectionViaFallback(e);
46: } else if (t instanceof HystrixTimeoutException) {
47: return handleTimeoutViaFallback();
48: } else if (t instanceof HystrixBadRequestException) {
49: return handleBadRequestByEmittingError(e);
50: } else {
51: /*
52: * Treat HystrixBadRequestException from ExecutionHook like a plain HystrixBadRequestException.
53: */
54: if (e instanceof HystrixBadRequestException) {
55: eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
56: return Observable.error(e);
57: }
58: 
59: return handleFailureViaFallback(e);
60: }
61: }
62: };
63: 
64: // TODO 【2008】【请求缓存】
65: final Action1<Notification<? super R>> setRequestContext = new Action1<Notification<? super R>>() {
66: @Override
67: public void call(Notification<? super R> rNotification){
68: setRequestContextIfNeeded(currentRequestContext);
69: }
70: };
71: 
72: Observable<R> execution;
73: if (properties.executionTimeoutEnabled().get()) {
74: execution = executeCommandWithSpecifiedIsolation(_cmd)
75: .lift(new HystrixObservableTimeoutOperator<R>(_cmd)); // 超时
76: } else {
77: execution = executeCommandWithSpecifiedIsolation(_cmd);
78: }
79: 
80: return execution.doOnNext(markEmits)
81: .doOnCompleted(markOnCompleted)
82: .onErrorResumeNext(handleFallback)
83: .doOnEach(setRequestContext);
84: }

 1: private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd){
 2: if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
 3: // mark that we are executing in a thread (even if we end up being rejected we still were a THREAD execution and not SEMAPHORE)
 4: return Observable.defer(new Func0<Observable<R>>() {
 5: @Override
 6: public Observable<R> call(){
 7: 
 8: // 标记 executionResult 执行已发生
 9: executionResult = executionResult.setExecutionOccurred();
 10: 
 11: // 设置 commandState 为 USER_CODE_EXECUTED
 12: if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
 13: return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
 14: }
 15: 
 16: // TODO 【2002】【metrics】
 17: metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD);
 18: 
 19: // TODO 【2009】【执行超时】
 20: if (isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT) {
 21: // the command timed out in the wrapping thread so we will return immediately
 22: // and not increment any of the counters below or other such logic
 23: return Observable.error(new RuntimeException("timed out before executing run()"));
 24: }
 25: 
 26: // 设置 线程状态 为 ThreadState.STARTED
 27: if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) {
 28: // TODO 【2002】【metrics】
 29: //we have not been unsubscribed, so should proceed
 30: HystrixCounters.incrementGlobalConcurrentThreads();
 31: threadPool.markThreadExecution();
 32: 
 33: // TODO 【2010】【endCurrentThreadExecutingCommand】
 34: // store the command that is being run
 35: endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
 36: 
 37: // 标记 executionResult 使用线程执行
 38: executionResult = executionResult.setExecutedInThread();
 39: /**
40: * If any of these hooks throw an exception, then it appears as if the actual execution threw an error
41: */
 42: try {
 43: // TODO 【2003】【HOOK】
 44: executionHook.onThreadStart(_cmd);
 45: executionHook.onRunStart(_cmd);
 46: executionHook.onExecutionStart(_cmd);
 47: 
 48: // 获得 执行Observable
 49: return getUserExecutionObservable(_cmd);
 50: } catch (Throwable ex) {
 51: return Observable.error(ex);
 52: }
 53: } else {
 54: //command has already been unsubscribed, so return immediately
 55: return Observable.empty();
 56: }
 57: }
 58: }).doOnTerminate(new Action0() {
 59: @Override
 60: public void call(){
 61: if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.TERMINAL)) {
 62: handleThreadEnd(_cmd);
 63: }
 64: if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.TERMINAL)) {
 65: //if it was never started and received terminal, then no need to clean up (I don't think this is possible)
 66: }
 67: //if it was unsubscribed, then other cleanup handled it
 68: }
 69: }).doOnUnsubscribe(new Action0() {
 70: @Override
 71: public void call(){
 72: if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.UNSUBSCRIBED)) {
 73: handleThreadEnd(_cmd);
 74: }
 75: if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.UNSUBSCRIBED)) {
 76: //if it was never started and was cancelled, then no need to clean up
 77: }
 78: //if it was terminal, then other cleanup handled it
 79: }
 80: }).subscribeOn(threadPool.getScheduler(new Func0<Boolean>() { // TODO 芋艿：Scheduler
 81: @Override
 82: public Boolean call(){
 83: return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;
 84: }
 85: }));
 86: } else {
 87: return Observable.defer(new Func0<Observable<R>>() {
 88: @Override
 89: public Observable<R> call(){
 90: // 标记 executionResult 执行已发生
 91: executionResult = executionResult.setExecutionOccurred();
 92: 
 93: // 设置 commandState 为 USER_CODE_EXECUTED
 94: if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
 95: return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
 96: }
 97: 
 98: // TODO 【2002】【metrics】
 99: metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.SEMAPHORE);
100: 
101: // TODO 【2010】【endCurrentThreadExecutingCommand】
102: // semaphore isolated
103: // store the command that is being run
104: endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
105: try {
106: // TODO 【2003】【HOOK】
107: executionHook.onRunStart(_cmd);
108: executionHook.onExecutionStart(_cmd);
109: 
110: // 获得 执行Observable
111: return getUserExecutionObservable(_cmd); //the getUserExecutionObservable method already wraps sync exceptions, so this shouldn't throw
112: } catch (Throwable ex) {
113: //If the above hooks throw, then use that as the result of the run method
114: return Observable.error(ex);
115: }
116: }
117: });
118: }
119: }

 1: private Observable<R> getUserExecutionObservable(final AbstractCommand<R> _cmd){
 2: Observable<R> userObservable;
 3: 
 4: try {
 5: userObservable = getExecutionObservable();
 6: } catch (Throwable ex) {
 7: // the run() method is a user provided implementation so can throw instead of using Observable.onError
 8: // so we catch it here and turn it into Observable.error
 9: userObservable = Observable.error(ex);
10: }
11: 
12: return userObservable
13: .lift(new ExecutionHookApplication(_cmd)) // TODO 【2003】【HOOK】
14: .lift(new DeprecatedOnRunHookApplication(_cmd)); // 已废弃
15: }

 1: @Override
 2: final protected Observable<R> getExecutionObservable(){
 3: return Observable.defer(new Func0<Observable<R>>() {
 4: @Override
 5: public Observable<R> call(){
 6: try {
 7: return Observable.just(run());
 8: } catch (Throwable ex) {
 9: return Observable.error(ex);
10: }
11: }
12: }).doOnSubscribe(new Action0() {
13: @Override
14: public void call(){
15: // 记录 执行线程
16: // Save thread on which we get subscribed so that we can interrupt it later if needed
17: executionThread.set(Thread.currentThread());
18: }
19: });
20: }
21: 
22: protected abstract R run() throws Exception;

protected enum CommandState {
 NOT_STARTED, OBSERVABLE_CHAIN_CREATED, USER_CODE_EXECUTED, UNSUBSCRIBED, TERMINAL
}

protected enum ThreadState {
 NOT_USING_THREAD, STARTED, UNSUBSCRIBED, TERMINAL
}