Spring Cloud 原始碼學習之 Hystrix 工作原理

Hystrix SpringCloud · 發表 2018-11-28 09:39:56

摘要：參考資訊與文中連結請點選閱讀原文，感興趣可移步PC端閱讀，移動端閱讀原始碼很不友好(截圖字太小，貼程式碼排版太亂)。本文學習了 Hystrix 工作原理及原始碼，關注點在整體處理流程，不涉及具體的實現細節。後續將逐漸寫Metrics收集、斷路器、隔離、請求快取等，有興趣可以關注奧。 ...

參考資訊與文中連結請點選閱讀原文，感興趣可移步PC端閱讀，移動端閱讀原始碼很不友好(截圖字太小，貼程式碼排版太亂)。

本文學習了 Hystrix 工作原理及原始碼，關注點在整體處理流程，不涉及具體的實現細節。後續將逐漸寫Metrics收集、斷路器、隔離、請求快取等，有興趣可以關注奧。

下面流程圖來源於 Hystrix Wiki，展現了 Hystrix 工作原理，官方 Wiki 中對每一步都做了詳細的描述，可以直接參考。

文中原始碼基於 Spring Cloud Finchley.SR1 、Spring Boot 2.0.6.RELEASE .

工作原理簡述

當需要完成某項任務時，通過 Hystrix 將任務包裹起來，交由 Hystrix 來完成任務，從而享受 Hystrix 帶來保護。這和古代鏢局生意有點類似，將任務委託給鏢局，以期安全完成任務。

上圖展示了 Hystrix 完成任務的處理流程，下面對1到9步驟進行簡述：

1.構建命令

Hystrix 提供了兩個Command， HystrixCommand 和 HystrixObservableCommand ，可以使用這兩個物件來包裹待執行的任務。

例如使用 @HystrixCommand 註解標記方法，Hystrix 將利用AOP自動將目標方法包裝成HystrixCommand來執行。

@HystrixCommand
public String hello() {
   ...
}

也可以繼承HystrixCommand或HystrixObservableCommand來建立Command，例如：

public class MyCommand extends HystrixCommand {
 
   public MyCommand(HystrixCommandGroupKey group) {
       super(group);
   }
 
   @Override
   protected Object run() throws Exception {
       // 需要做的事情及需要返回的結果
       return null;
   }
}

任務委託給 Hystrix 後，Hystrix 可以應用自己的一系列保護機制，在執行使用者任務的各節點(執行前、執行後、異常、超時等)做一系列的事情。

2.執行命令

有四種方式執行command。

R execute()：同步執行，從依賴服務得到單一結果物件
Future queue() ：非同步執行，返回一個 Future 以便獲取執行結果，也是單一結果物件
Observable observe() ：hot observable，建立Observable後會訂閱Observable，可以返回多個結果
Observable toObservable() ：cold observable，返回一個Observable，只有訂閱時才會執行，可以返回多個結果

execute()的實現為 queue().get() ； queue() 的實現為 toObservable().toBlocking().toFuture() 。

最後Obserable都由toObservable()來建立，本文的主要內容就是toObservable()。

// 利用queue()拿到Future, 執行 get()同步等待拿到執行結果
public R execute() {
   ...
   return queue().get();
}
 
// 利用toObservable()得到Observable最後轉成Future
public Future<R> queue() {
   final Future<R> delegate = toObservable().toBlocking().toFuture();
   ...
}
 
// 利用toObservable()得到Observable並直接訂閱它,立即執行命令
public Observable<R> observe() {
   ReplaySubject<R> subject = ReplaySubject.create();
   final Subscription sourceSubscription = toObservable().subscribe(subject);
   ...
}

3.檢查快取

第3到9步驟構成了 Hystrix 的保護能力，通過這一些列步驟來執行任務，從而起到保護作用。

如果啟用了 Hystrix Cache，任務執行前將先判斷是否有相同命令執行的快取。如果有則直接返回快取的結果；如果沒有快取的結果，但啟動了快取，將快取本次執行結果以供後續使用。

4.檢查斷路器是否開啟

斷路器(circuit-breaker)和保險絲類似，保險絲在發生危險時將會燒斷以保護電路，而斷路器可以在達到我們設定的閥值時觸發短路(比如請求失敗率達到50%)，拒絕執行任何請求。

如果斷路器被開啟，Hystrix 將不會執行命令，直接進入Fallback處理邏輯。

5.檢查執行緒池/訊號量情況

Hystrix 隔離方式有執行緒池隔離和訊號量隔離。當使用Hystrix執行緒池時，Hystrix 預設為每個依賴服務分配10個執行緒，當10個執行緒都繁忙時，將拒絕執行命令。訊號量同理。

6.執行具體的任務

通過 HystrixObservableCommand.construct() 或者 HystrixCommand.run() 來執行使用者真正的任務。

7.計算鏈路健康情況

每次開始執行command、結束執行command以及發生異常等情況時，都會記錄執行情況，例如：成功、失敗、拒絕以及超時等情況，會定期處理這些資料，再根據設定的條件來判斷是否開啟斷路器。

8.命令失敗時執行 Fallback 邏輯

在命令失敗時執行使用者指定的 Fallback 邏輯。上圖中的斷路、執行緒池拒絕、訊號量拒絕、執行執行、執行超時都會進入 Fallback 處理。

9.返回執行結果

原始結果將以Observable形式返回，在返回給使用者之前，會根據呼叫方式的不同做一些處理。

下面是 Hystrix Return flow。

原始碼學習

小故事

由於最終入口都是 toObservable() ，就從 AbstractCommand的 Observable<R> toObservable() 方法開始。

Hystrix 使用觀察者模式， Observable 即被觀察者，被觀察者狀態變更時，觀察者可以做出各項響應。舉個例子：大廳中一位演講者正在分享，廳中有觀眾和工作人員，可能發生如下事情：

被觀察者 事件            觀察者
-----------------------------------
演講者 分享到精彩處 -> 觀眾鼓掌
演講者 講的口乾舌燥 -> 工作人員遞上一瓶水
演講者 放出自己的二維碼 -> 觀眾掃描

因為 Hystrix 基於RxJava，RxJava 初次看會比較複雜。為了便於下文理解，可以將Observable理解為資料來源、資料發射器，上面例子中，演講者各種行為都可以抽象為資料來源在發射資料，而各種接收者可以做出各種響應。

toObservable()

toObservable()主要原始碼如下：

public Observable<R> toObservable() {
   final AbstractCommand<R> _cmd = this;
   // 命令執行結束後的清理者
   final Action0 terminateCommandCleanup = new Action0() {...};
   // 取消訂閱時處理者
   final Action0 unsubscribeCommandCleanup = new Action0() {...};
   // 重點：Hystrix 核心邏輯: 斷路器、隔離
   final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {...};
   // 發射資料(OnNext表示發射資料)時的Hook
   final Func1<R, R> wrapWithAllOnNextHooks = new Func1<R, R>() {...};
   // 命令執行完成的Hook
   final Action0 fireOnCompletedHook = new Action0() {...};
 
   // 通過Observable.defer()建立一個Observable
   return Observable.defer(new Func0<Observable<R>>() {
       @Override
       public Observable<R> call() {
           final boolean requestCacheEnabled = isRequestCachingEnabled();
           final String cacheKey = getCacheKey();
 
           // 首先嚐試從請求快取中獲取結果
           if (requestCacheEnabled) {
               HystrixCommandResponseFromCache<R> fromCache = (HystrixCommandResponseFromCache<R>) requestCache.get(cacheKey);
               if (fromCache != null) {
                   isResponseFromCache = true;
                   return handleRequestCacheHitAndEmitValues(fromCache, _cmd);
               }
           }
 
           // 使用上面的Func0：applyHystrixSemantics 來建立Observable
           Observable<R> hystrixObservable =
                   Observable.defer(applyHystrixSemantics)
                           .map(wrapWithAllOnNextHooks);
 
           Observable<R> afterCache;
 
           // 如果啟用請求快取，將Observable包裝成HystrixCachedObservable並進行相關處理
           if (requestCacheEnabled && cacheKey != null) {
               HystrixCachedObservable<R> toCache = HystrixCachedObservable.from(hystrixObservable, _cmd);
               ...
           } else {
               afterCache = hystrixObservable;
           }
 
           // 返回Observable
           return afterCache
                   .doOnTerminate(terminateCommandCleanup)  
                   .doOnUnsubscribe(unsubscribeCommandCleanup)
                   .doOnCompleted(fireOnCompletedHook);
       }
   });
}

上面的程式碼可以換種思維方式來理解。平時開發時都是下面這種模式，按順序不斷的做事情，是一個很好的執行者。

public void methodA{
   try {
       // 1. 做第一件事情
       // 2. 呼叫methodB()做第二件事情
       // 3. 做第三件事情
       ...
   } catch (Exception e) {
       // 處理錯誤
   } finally {
       // 最後一定要做的事情
   }
}

用一張圖來看 toObservable() 方法。這種方式是“軍師型”，排兵佈陣，先創造了各個處理者，然後創造被觀察者，再設定Observable發生各種情況時由誰來處理，完全掌控全域性。

解釋下Action0、Func1這種物件。Action、Func和Runnable、Callable類似，是一個可以被執行的實體。Action沒有返回值，Action0…ActionN表示有0..N個引數，Action0就表示沒有引數；Func有返值，0..N一樣表示引數。

public interface Action0 extends Action {
   void call();
}
public interface Func1<T, R> extends Function {
   R call(T t);
}

下面用核心的 applyHystrixSemantics 來闡述一下。

// applyHystrixSemantics 是一個Func0(理解為執行實體或處理者),表示沒有引數,返回值是Observable。
final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {
   // Func0 做的事情如下
   @Override
   public Observable<R> call() {
       // 如果未訂閱,返回一個"啞炮" Observable, 即一個不會發射任何資料的Observable
       if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {
           return Observable.never();
       }
       // 呼叫applyHystrixSemantics()來建立Observable
       return applyHystrixSemantics(_cmd);
   }
};

因此，當執行Func0: applyHystrixSemantics時，可以得到一個Observable。 toObservable() 大量程式碼在準備處理者(觀察者)，實際使用時是方法最後的 Observable.defer(new Func0<observable >(){…}</observable

Observable.defer

defer譯為延遲，表示演講者會等有觀眾來時才開始分享。Observable.defer() 就是說：必須有觀察者訂閱時， Observable 才開始發射資料。而defer()的引數是個Func0，是一個會返回Observable的執行實體。下面看看defer()：

return Observable.defer(new Func0<Observable<R>>() {
   @Override
   public Observable<R> call() {
       // 再一次使用Observable.defer()技能，這次用的是applyHystrixSemantics這個Func0
       Observable<R> hystrixObservable =
               Observable.defer(applyHystrixSemantics)
                       .map(wrapWithAllOnNextHooks);
       ... // 此處忽略了請求快取處理,上面已有提及
       Observable<R> afterCache;
       ...
       // 為Observable繫結幾個特定事件的處理者,這都是上門建立的Action0
       return afterCache
               .doOnTerminate(terminateCommandCleanup)
               .doOnUnsubscribe(unsubscribeCommandCleanup)
               .doOnCompleted(fireOnCompletedHook);
   }
});

applyHystrixSemantics()

接著看applyHystrixSemantics這個Func0，Func0的call()中呼叫的是applyHystrixSemantics()函式。

// Semantics 譯為語義, 應用Hystrix語義很拗口，其實就是應用Hystrix的斷路器、隔離特性
private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
   // 原始碼中有很多executionHook、eventNotifier的操作，這是Hystrix拓展性的一種體現。這裡面啥事也沒做，留了個口子，開發人員可以拓展
   executionHook.onStart(_cmd);
 
   // 判斷斷路器是否開啟
   if (circuitBreaker.attemptExecution()) {
       // 獲取執行訊號
       final TryableSemaphore executionSemaphore = getExecutionSemaphore();
       final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);
       final Action0 singleSemaphoreRelease = new Action0() {...};
       final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {...};
 
       // 判斷是否訊號量拒絕
       if (executionSemaphore.tryAcquire()) {
           try {
               // 重點：處理隔離策略和Fallback策略
               return executeCommandAndObserve(_cmd)
                       .doOnError(markExceptionThrown)
                       .doOnTerminate(singleSemaphoreRelease)
                       .doOnUnsubscribe(singleSemaphoreRelease);
           } catch (RuntimeException e) {
               return Observable.error(e);
           }
       } else {
           return handleSemaphoreRejectionViaFallback();
       }
   }
   // 開啟了斷路器,執行Fallback
   else {
       return handleShortCircuitViaFallback();
   }
}

executeCommandAndObserve()

下面看 executeCommandAndObserve() 方法，處理隔離策略和各種Fallback.

private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
   final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
 
   final Action1<R> markEmits = new Action1<R>() {...};
   final Action0 markOnCompleted = new Action0() {...};
 
   // 利用Func1獲取處理Fallback的 Observable
   final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
       @Override
       public Observable<R> call(Throwable t) {
           circuitBreaker.markNonSuccess();
           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 {
               ...
               return handleFailureViaFallback(e);
           }
       }
   };
 
   final Action1<Notification<? super R>> setRequestContext ...
 
   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)
           // 繫結Fallback的處理者
           .onErrorResumeNext(handleFallback)
           .doOnEach(setRequestContext);
}

executeCommandWithSpecifiedIsolation()

接著看隔離特性的處理：executeCommandWithSpecifiedIsolation()

private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
   // 執行緒池隔離
   if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
       // 再次使用 Observable.defer(), 通過執行Func0來得到Observable
       return Observable.defer(new Func0<Observable<R>>() {
           @Override
           public Observable<R> call() {
               // 收集metric資訊
               metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD);
               ...
               try {
                     ... // 獲取真正的使用者Task
                   return getUserExecutionObservable(_cmd);
               } catch (Throwable ex) {
                   return Observable.error(ex);
               }
               ...
           }
           // 繫結各種處理者
       }).doOnTerminate(new Action0() {...})
           .doOnUnsubscribe(new Action0() {...})
           // 繫結超時處理者
           .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>>() {...}
   }
}

getUserExecutionObservable()就不接著寫了，可以自己看下，就是拿到使用者真正要執行的任務。這個任務就是這樣被Hystrix包裹著，置於層層防護之下。

倒過來看

上面方法層層呼叫，倒過來看，就是先建立一個Observable，然後繫結各種事件對應的處理者，如下圖：

各類doOnXXXX，表示發生XXX事件時做什麼事情。