spring-cloud-netflix-hystrix 原理源碼分析：
本文主要針對 spring-cloud-dependencies Hoxton.SR4版本, spring-cloud-starter-netflix-hystrix 源碼的解析。
對於未接觸過 hystrix 的小伙伴可以參考 https://www.cnblogs.com/wuzhenzhao/p/9473073.html 進行一些基礎知識的了解。
本文主要從以下幾個點來分析：
- 手寫實現簡易版 Hystrix 體驗。
- RXJava 基礎知識體驗。
- Hystrix 源碼流程分析。
手寫實現簡易版 Hystrix 體驗：
繼上文的博客鏈接，我們知道了Hystrix 提供了多種方式讓我們實現服務降級。我們可以通過注解 @HystrixCommand、或者繼承 HystrixCommand 來實現降級，以及一些請求合並等操作。
我們需要知道的是，當我們采用 @HystrixCommand 注解來實現服務降級，在Hystrix 的內部是采用 AOP 的方式進行攔截處理請求的，我們這里就先來實現一下簡易版的 Hystrix 來體會一下，主要分為以下步驟
- 定義自己的@HystrixCommand 注解。
- 實現攔截請求的處理邏輯。
- 測試調用。
1.自定義注解 @WuzzHystrixCommand
@Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @Documented public @interface WuzzHystrixCommand { /** * 默認超時時間 * * @return */
int timeout() default 1000; /** * 回退方法 * * @return */ String fallback() default ""; }
2.編寫切面類，實現簡易的邏輯處理
@Component @Aspect public class WuzzHystrixCommandAspect { //線程池的處理，基於這個線程池的處理統計可以達到 THREAD 資源限流
ExecutorService executorService = Executors.newFixedThreadPool(10); //注解切點
@Pointcut(value = "@annotation(com.wuzz.demo.custom.hystrix.WuzzHystrixCommand)") public void pointCut() { } //環繞通知
@Around(value = "pointCut()&&@annotation(hystrixCommand)") public Object doPointCut(ProceedingJoinPoint joinPoint, WuzzHystrixCommand hystrixCommand) throws InterruptedException, ExecutionException, TimeoutException, NoSuchMethodException, InvocationTargetException, IllegalAccessException { int timeout = hystrixCommand.timeout(); //前置的判斷邏輯
Future future = executorService.submit(() -> { try { return joinPoint.proceed(); //執行目標方法
} catch (Throwable throwable) { throwable.printStackTrace(); } return null; }); Object result; try {// 使用 future 來實現超時
result = future.get(timeout, TimeUnit.MILLISECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { e.printStackTrace(); future.cancel(true); // ？
if (StringUtils.isBlank(hystrixCommand.fallback())) { throw e; } //調用fallback
result = invokeFallback(joinPoint, hystrixCommand.fallback()); } return result; } private Object invokeFallback(ProceedingJoinPoint joinPoint, String fallback) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException { //獲取被代理的方法的參數和Method
MethodSignature signature = (MethodSignature) joinPoint.getSignature(); Method method = signature.getMethod(); Class<?>[] parameterTypes = method.getParameterTypes(); //得到fallback方法
try { Method fallbackMethod = joinPoint.getTarget().getClass().getMethod(fallback, parameterTypes); fallbackMethod.setAccessible(true); //完成反射調用
return fallbackMethod.invoke(joinPoint.getTarget(), joinPoint.getArgs()); } catch (Exception e) { e.printStackTrace(); throw e; } } }
3. 編寫測試，調用：
@WuzzHystrixCommand(fallback = "customFallback", timeout = 3000) @GetMapping("/custom/hystrix/test") public String test() { Map map = new HashMap<>(); map.put("id", 666); return restTemplate.getForObject(REST_URL_PREFIX + "/hello?id={id}", String.class, map); } public String customFallback() { return "custom 請求被降級"; }
正常得調用是沒有問題的，這個時候我們把服務提供方的服務接口里 sleep 3秒來模仿調用超時，在訪問接口：
相信小伙伴們有了一些心得了，只不過Hystrix里面得實現是很復雜的 ，沒有我們這么簡單。
RXJava 基礎知識體驗：
上文的博文連接中也講到了 RXJava的簡單例子，這里由於馬上我們要去看 Hystrix的源碼了，我們這里寫一個類似於源碼中的例子，這樣來幫助我們更容易理解。
public class RxJavaDemo { // ReactiveX Java 響應式編程框架(android） // Java stream() java8 //觀察者模式
public static void main(String[] args) throws ExecutionException, InterruptedException { final String[] datas = new String[]{"登錄"}; final Action0 onComplated = new Action0() { @Override public void call() { System.out.println("on Complated"); } }; //老師（被觀察者）
Observable<String> observable = Observable.defer(new Func0<Observable<String>>() { @Override public Observable<String> call() { Observable observable1 = Observable.from(datas); return observable1.doOnCompleted(onComplated); } }); //學生(觀察者)
Observer observer = new Observer() { @Override public void onCompleted() { System.out.println("Observer: onCompleted"); } @Override public void onError(Throwable throwable) { System.out.println("Observer: onError"); } @Override public void onNext(Object o) { System.out.println("on Next:" + o); } }; // observable.subscribe(observer); //建立訂閱關系
String s = observable.toBlocking().toFuture().get();//建立訂閱關系
System.out.println(s); } }
寫這個例子的目的主要是想說明，我們可能對於 RXJava 的 API可能不是很熟悉，但是我們一定要知道對於 Observable 實例來說， call 方法才是關鍵，而 observable.toBlocking().toFuture().get() 是用於獲取執行結果的。在 Hystrix的源碼中能看到。了解一下，我們直接進入Hystrix的源碼
Hystrix 源碼流程分析：
需要注意的是，Hystrix用到了RxJava這個框架，它是一個響應式編程框架，在Android里面用得比較多，所以很多同學對它不是很了解。如果不了解的話，看Hystrix的源碼就會有點困難。
Hystrix的數據統計是采用的滑動窗口，關於滑動窗口我這里就不深入研究了，又興趣的同學可以參考我另外一篇博客， Sentinel 限流原理 進行了解，也可以直接訪問 滑動窗口在線演示地址。
Hystrix熔斷的@HystrixCommand注解，是通過HystrixCommandAspect這個切面來處理的。其中我們關注@Around注解聲明的方法，它針對於請求合並，以及降級的注解進行代理。這里我們重點針對HystrixCommand這個注解進行詳細分析。
@Aspect public class HystrixCommandAspect { private static final Map<HystrixPointcutType, MetaHolderFactory> META_HOLDER_FACTORY_MAP; static { META_HOLDER_FACTORY_MAP = ImmutableMap.<HystrixPointcutType, MetaHolderFactory>builder() .put(HystrixPointcutType.COMMAND, new CommandMetaHolderFactory()) .put(HystrixPointcutType.COLLAPSER, new CollapserMetaHolderFactory()) .build(); } // 熔斷降級切點
@Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)") public void hystrixCommandAnnotationPointcut() { } // 請求合並切點
@Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)") public void hystrixCollapserAnnotationPointcut() { } // 環繞通知
@Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()") public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable { // 獲取目標方法信息
Method method = getMethodFromTarget(joinPoint); Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint); if (method.isAnnotationPresent(HystrixCommand.class) && method.isAnnotationPresent(HystrixCollapser.class)) { throw new IllegalStateException("method cannot be annotated with HystrixCommand and HystrixCollapser " +
"annotations at the same time"); } MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method)); // 獲取元數據，比如調用方法，HystrixProperty注解數據、方法參數等
MetaHolder metaHolder = metaHolderFactory.create(joinPoint); //獲取調用者，它持有一個命令對象，並且可以在合適的時候通過這個命令對象完成具體的業務邏輯
HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder); ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ? metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType(); Object result; try { //是否是響應式的（由於我們這些都是同步的會走這個邏輯）
if (!metaHolder.isObservable()) { result = CommandExecutor.execute(invokable, executionType, metaHolder); } else { result = executeObservable(invokable, executionType, metaHolder); } } catch (HystrixBadRequestException e) { throw e.getCause(); } catch (HystrixRuntimeException e) { throw hystrixRuntimeExceptionToThrowable(metaHolder, e); } return result; } // ......
}
然后進入 CommandExecutor#execute 方法這個方法主要用來執行命令，從代碼中可以看出這里有三個執行類型，分別是同步、異步、以及響應式。其中，響應式又分為Cold Observable（observable.toObservable()） 和 HotObservable（observable.observe()）默認的executionType=SYNCHRONOUS ，同步請求。
- execute():同步執行，返回一個單一的對象結果，發生錯誤時拋出異常。
- queue():異步執行，返回一個 Future 對象，包含着執行結束后返回的單一結果。
- observe():這個方法返回一個 Observable 對象，它代表操作的多個結果，但是已經被訂閱者消費掉了。
- toObservable():這個方法返回一個 Observable 對象，它代表操作的多個結果，需要咱們自己手動訂閱並消費掉。
接着調用HystrixCommand.execute()方法，這個方法中，首先調用queue()，這個方法會返回一個future對象。
public R execute() { try { return queue().get(); } catch (Exception e) { throw Exceptions.sneakyThrow(decomposeException(e)); } }
queue這個方法中，返回了一個Future對象，這個future對象的實現是f，f是以匿名內部類，它是Java.util.concurrent中定一個的一個異步帶返回值對象。當調用queue().get()方法時，最終是委派給了delegate.get 方法。
public Future<R> queue() { /* * The Future returned by Observable.toBlocking().toFuture() does not implement the * interruption of the execution thread when the "mayInterrupt" flag of Future.cancel(boolean) is set to true; * thus, to comply with the contract of Future, we must wrap around it. */ final Future<R> delegate = toObservable().toBlocking().toFuture(); final Future<R> f = new Future<R>() { @Override public boolean cancel(boolean mayInterruptIfRunning) { // ......省略代碼
@Override public R get() throws InterruptedException, ExecutionException { return delegate.get(); } @Override public R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return delegate.get(timeout, unit); } }; // ...省略代碼
return f; }
在上述代碼中，重點來了，構建了一個 java.util.concurrent.Future ，然后調用 get的時候委派給 delegate，而 delegate來自於 toObservable().toBlocking().toFuture(); 這正是我們上面例子里面得代碼。所以我們現在的重點應該放在 toObservable() 方法中：
AbstractCommand.toObservable ：通過Observable定義一個被觀察者，這個被觀察者會被toObservable().toBlocking().toFuture() ，實際上就是返回可獲得 run() 抽象方法執行結果的Future 。 run() 方法由子類實現，執行正常的業務邏輯。在下面這段代碼中，當存在subscriber時，便會調用Func0#call() 方法，而這個subscriber是在 toBlocking() 中被訂閱的。到這還是我們上面的例子里面的代碼。該方法主要做了以下幾件事：
-
創建一些命令供后續的回調使用
-
調用 isRequestCachingEnabled(); 判斷請求結果緩存功能是否開啟，如果開啟並且命中了緩存，則會以Observable形式返回一個緩存結果
-
創建執行命令的Observable： hystrixObservable，
-
當緩存處於開啟狀態並且沒有命中緩存時，則創建一個“訂閱了執行命令的Observable”：HystrixCommandResponseFromCache
-
創建存儲到緩存的Observable： HystrixCachedObservable當緩存特性沒有開啟時，則返回執行命令的Observable。
-
將toCache添加到緩存中，返回獲取緩存的Observable：fromCache
-
如果添加失敗: fromCache!=null, 則調用 toCache.unsubscribe() 方法，取消HystrixCachedObservable 的訂閱
-
如果添加成功，則調用 toCache.toObservable(); 獲得緩存Observable
-
-
當緩存特性沒有開啟時，則返回執行命令的Observable。
public Observable<R> toObservable() { final AbstractCommand<R> _cmd = this; //會在Observable結束前觸發回調該call方法，無論是正常還是異常終止 //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
} } }; //取消訂閱時的監聽會進行回調該 call方法 //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
} } }; final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() { @Override public Observable<R> call() { if (commandState.get().equals(CommandState.UNSUBSCRIBED)) { return Observable.never(); // 立即終止整個流程。
}//返回執行命令的Observable
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>>() { @Override public Observable<R> call() { /* this is a stateful object so can only be used once */
// CAS保證命令只執行一次
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); } } // 緩存開關，緩存KEY（這個是Hystrix中請求緩存功能，hystrix支持將一個請求結果緩存起來， // 下一個具有相同key的請求將直接從緩存中取出結果，減少請求開銷）
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); } } // 聲明執行命令的Observable
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 //會在Observable結束前觸發回調，無論是正常還是異常終止
.doOnTerminate(terminateCommandCleanup) // perform cleanup once (either on normal terminal state (this line), or unsubscribe (next line)) //取消訂閱時的監聽
.doOnUnsubscribe(unsubscribeCommandCleanup) // perform cleanup once //Observable正常終止時的監聽
.doOnCompleted(fireOnCompletedHook); } }); }
所以在 AbstractCommand#toObservable 方法里，我們只需要看這個返回的 Observable 對象的 call 方法即可，而在這里 默認沒有開啟緩存的話就是 ：
Observable<R> hystrixObservable = Observable.defer(applyHystrixSemantics) .map(wrapWithAllOnNextHooks);
那么我們主要來看 applyHystrixSemantics ，在該方法上面定義了
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); } };
假設緩存特性未開啟或者未命中緩存，那么代碼將執行 applyHystrixSemantics 。傳入的_cmd是一個GenericCommand(可以斷點看看)，最終執行這個command中的run方法，本質就是完成對queryOrder方法的代理。
circuitBreaker.allowRequest() 如果為true，表示當前不處於熔斷狀態，正常執行，否則，調用 handleShortCircuitViaFallback 實現服務降級，如果我們配置了fallback方法，則會獲得我們配置的fallback執行。執行路徑為 : handleShortCircuitViaFallback ->getFallbackOrThrowException ->getFallbackObservable->HystrixCommand.getFallbackObservable->GenericCommand.getFallback();
如果當前hystrix處於未熔斷狀態，則
- getExecutionSemaphore 判斷當前策略是否為信號量（TryableSemaphoreNoOp/TryableSemaphoreActual），如果是，則調用 tryAcquire 來獲取信號量。如果當前信號量滿了，則調用 handleSemaphoreRejectionViaFallback 方法。
- 調用 executeCommandAndObserve 獲取命令執行Observable。
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
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); } }; // 是否開啟信號量資源隔離，未配置走 com.netflix.hystrix.AbstractCommand.TryableSemaphoreNoOp#tryAcquire 默認返回通過
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(); } }
我們縣來看一下執行失敗進入降級的邏輯，這里我們直接進入到 HystrixCommand#getFallbackObservable
@Override final protected Observable<R> getFallbackObservable() { return Observable.defer(new Func0<Observable<R>>() { @Override public Observable<R> call() { try { return Observable.just(getFallback()); } catch (Throwable ex) { return Observable.error(ex); } } }); }
這里的 getFallback 我們應該熟悉了，因為通過集成 HystrixCommand 類來實現熔斷降級的時候我們重寫了這個方法，而通過注解的話是通過 GenericCommand進行代理實現得，我們Debug一下，看看該類的 getFallback 方法做了什么：
可以發現他拿到了我們配置在注解上的方法，這一點是不是跟上文的手寫是一個道理呢？ 然后進行調用獲取結果返回。
好了，回到 AbstractCommand#applyHystrixSemantics ，接下去我們按照正常邏輯走到 AbstractCommand#executeCommandAndObserve，主要做了以下三件事情
- 定義不同的回調，doOnNext、doOnCompleted、onErrorResumeNext、doOnEach。
- 調用executeCommandWithSpecifiedIsolation獲得執行命令的Observable
- 若執行命令超時特性開啟，調用 Observable.lift 方法實現執行命令超時功能。
private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) { final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread(); // Action和Func都是定義的一個動作，Action是無返回值，Func是有返回值 // doOnNext中的回調。即命令執行之前執行的操作
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(); } } }; // doOnCompleted中的回調。命令執行完畢后執行的操作
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(); } } }; // onErrorResumeNext中的回調。命令執行失敗后的回退邏輯
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); } } }; // doOnEach中的回調。`Observable`每發射一個數據都會執行這個回調，設置請求上下文
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:這個方法首先是根據當前不同的資源隔離策略執行不同的邏輯，THREAD、SEMAPHORE:
private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) { // 是否開啟 THREAD 資源隔離降級
if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) { // ......省略代碼 } 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); // semaphore isolated // store the command that is being run
endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey()); try { executionHook.onRunStart(_cmd); executionHook.onExecutionStart(_cmd); // 真正的執行
return getUserExecutionObservable(_cmd); //the getUserExecutionObservable method already wraps sync exceptions, so this shouldn't throw
} catch (Throwable ex) { //If the above hooks throw, then use that as the result of the run method
return Observable.error(ex); } } }); } }
這里就不展開實現細節，我們直接看執行的方法 getUserExecutionObservable 。然后會執行 HystrixCommand#getExecutionObservable
@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()); } }); }
又看到熟悉的代碼 ，這個 run() 方法在通過集成 HystrixCommand 類來實現熔斷降級的時候我們重寫了這個方法，是真正的執行方法。
這里最終調用的是run方法，通過Observable.just， just是RxJava中的一個操作符，它可以接受一個或者多個參數來創建一個Observable對象。而這個run（）方法是一個抽象方法，在HystrixCommand中並沒有實現，而是在子類中實現，而此時傳遞的cmd=GenricCommand正好實現了HystrixCommand，重寫了run方法。
@Override protected Object run() throws Exception { LOGGER.debug("execute command: {}", getCommandKey().name()); return process(new Action() { @Override Object execute() { return getCommandAction().execute(getExecutionType()); } }); }
大家有沒有發現，這里的實現和我們前面自定義的 HystrixCommandService 實現是一樣的，同樣是集成HystrixCommand，重寫run方法。這里也是如此。
- 首先調用 getCommandAction() 方法獲取 CommandAction ，我們的示例中獲取到的是MethodExecutionAction 。
- 然后調用 MethodExecutionAction.execute 方法，傳入 ExecutionType 參數，我們的示例中傳入的是 ExecutionType.SYNCHRONOUS 。
拿到我們的真實方法進行調用返回。下面附上整個過程的流程圖：
