• Status 列舉類，斷路器的三種狀態。

• status 屬性，斷路器的狀態。

• circuitOpened 屬性，斷路器開啟，即狀態變成 OPEN 的時間。

• activeSubscription 屬性，基於 Hystrix Metrics 對請求量統計 Observable 的訂閱，在 「4.2 #subscribeToStream()」 詳細解析。

4.2 #subscribeToStream()

#subscribeToStream() 方法，向 Hystrix Metrics 對請求量統計 Observable 的發起訂閱。程式碼如下 ：

private Subscription subscribeToStream() {
  1: private Subscription subscribeToStream() {
  2:     /*
  3:      * This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream
  4:      */
  5:     return metrics.getHealthCountsStream()
  6:             .observe()
  7:             .subscribe(new Subscriber<HealthCounts>() {
  8:                 @Override
  9:                 public void onCompleted() {
 10: 
 11:                 }
 12: 
 13:                 @Override
 14:                 public void onError(Throwable e) {
 15: 
 16:                 }
 17: 
 18:                 @Override
 19:                 public void onNext(HealthCounts hc) {
 20:                     System.out.println("totalRequests" + hc.getTotalRequests()); // 芋艿，用於除錯
 21:                     // check if we are past the statisticalWindowVolumeThreshold
 22:                     if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
 23:                         // we are not past the minimum volume threshold for the stat window,
 24:                         // so no change to circuit status.
 25:                         // if it was CLOSED, it stays CLOSED
 26:                         // if it was half-open, we need to wait for a successful command execution
 27:                         // if it was open, we need to wait for sleep window to elapse
 28:                     } else {
 29:                         if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
 30:                             //we are not past the minimum error threshold for the stat window,
 31:                             // so no change to circuit status.
 32:                             // if it was CLOSED, it stays CLOSED
 33:                             // if it was half-open, we need to wait for a successful command execution
 34:                             // if it was open, we need to wait for sleep window to elapse
 35:                         } else {
 36:                             // our failure rate is too high, we need to set the state to OPEN
 37:                             if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
 38:                                 circuitOpened.set(System.currentTimeMillis());
 39:                             }
 40:                         }
 41:                     }
 42:                 }
 43:             });
 44: }

• 第 5 至 7 行 ：向 Hystrix Metrics 對請求量統計 Observable 的發起訂閱。這裡的 Observable 基於 RxJava Window 運運算元。

  FROM 《ReactiveX檔案中文翻譯》「Window」
  定期將來自原始 Observable 的資料分解為一個 Observable 視窗，發射這些視窗，而不是每次發射一項資料

• 簡單來說，固定間隔， #onNext() 方法將不斷被呼叫，每次計算斷路器的狀態。

• 第 22 行 ：判斷週期( 可配， HystrixCommandProperties.default_metricsRollingStatisticalWindow=10000ms )內，總請求數超過一定量( 可配， HystrixCommandProperties.circuitBreakerRequestVolumeThreshold=20 ) 。

• 這裡要註意下，請求次數統計的是週期內，超過週期的不計算在內。例如說， 00:00 內發起了 N 個請求， 00:11 不計算這 N 個請求。

• 第 29 行 ：錯誤請求佔總請求數超過一定比例( 可配， HystrixCommandProperties.circuitBreakerErrorThresholdPercentage=50% ) 。

• 第 37 至 39 行 ：滿足斷路器開啟條件，CAS 修改狀態( CLOSED=>OPEN )，並設定開啟時間( circuitOpened ) 。

• 【補充】第 5 至 7 行 ：? 怕寫在上面，大家有壓力。Hystrix Metrics 對請求量統計 Observable 使用了兩種 RxJava Window 運運算元 ：

• Observable#window(timespan, unit) 方法，固定週期( 可配， HystrixCommandProperties.metricsHealthSnapshotIntervalInMilliseconds=500ms )，發射 Observable 視窗。點選 BucketedCounterStream 構造方法 檢視呼叫處的程式碼。

• Observable#window(count, skip) 方法，每發射一次( skip) Observable 忽略 count ( 可配， HystrixCommandProperties.circuitBreakerRequestVolumeThreshold=20 ) 個資料項。為什麼？答案在第 22 行的程式碼，週期內達到一定請求量是斷路器開啟的一個條件。點選 BucketedRollingCounterStream 構造方法 檢視呼叫處的程式碼。

目前該方法有兩處呼叫 ：

• 「4.1 構造方法」，在建立 HystrixCircuitBreakerImpl 時，向 Hystrix Metrics 對請求量統計 Observable 的發起訂閱。固定間隔，計算斷路器是否要關閉( CLOSE )。

• 「4.4 #markSuccess()」，清空 Hystrix Metrics 對請求量統計 Observable 的統計資訊，取消原有訂閱，併發起新的訂閱。

4.3 #attemptExecution()

如下是 AbstractCommand#applyHystrixSemantics(_cmd) 方法，對 HystrixCircuitBreakerImpl#attemptExecution 方法的呼叫的程式碼 ：

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {
    // ...  省略無關程式碼
   /* determine if we're allowed to execute */
   if (circuitBreaker.attemptExecution()) {
        // 執行【正常邏輯】
   } else {
        // 執行【回退邏輯】
   }
}

• 使用 HystrixCircuitBreakerImpl#attemptExecution 方法，判斷是否可以執行正常邏輯。

---

#attemptExecution 方法，程式碼如下 ：

  1: @Override
  2: public boolean attemptExecution() {
  3:     // 強制 開啟
  4:     if (properties.circuitBreakerForceOpen().get()) {
  5:         return false;
  6:     }
  7:     // 強制 關閉
  8:     if (properties.circuitBreakerForceClosed().get()) {
  9:         return true;
 10:     }
 11:     // 開啟時間為空
 12:     if (circuitOpened.get() == -1) {
 13:         return true;
 14:     } else {
 15:         // 滿足間隔嘗試斷路器時間
 16:         if (isAfterSleepWindow()) {
 17:             //only the first request after sleep window should execute
 18:             //if the executing command succeeds, the status will transition to CLOSED
 19:             //if the executing command fails, the status will transition to OPEN
 20:             //if the executing command gets unsubscribed, the status will transition to OPEN
 21:             if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
 22:                 return true;
 23:             } else {
 24:                 return false;
 25:             }
 26:         } else {
 27:             return false;
 28:         }
 29:     }
 30: }

• 第 4 至 6 行 ：當 HystrixCommandProperties.circuitBreakerForceOpen=true ( 預設值 ： false) 時，即斷路器強制開啟，傳回 false 。當該配置接入配置中心後，可以動態實現開啟熔斷。為什麼會有該配置？當 HystrixCircuitBreaker 建立完成後，無法動態切換 NoOpCircuitBreaker 和 HystrixCircuitBreakerImpl ，透過該配置以實現類似效果。

• 第 8 至 10 行 ：當 HystrixCommandProperties.circuitBreakerForceClose=true ( 預設值 ： false) 時，即斷路器強制關閉，傳回 true 。當該配置接入配置中心後，可以動態實現關閉熔斷。為什麼會有該配置？當 HystrixCircuitBreaker 建立完成後，無法動態切換 NoOpCircuitBreaker 和 HystrixCircuitBreakerImpl ，透過該配置以實現類似效果。

• 第 12 至 13 行 ：斷路器開啟時間( circuitOpened ) 為"空"，傳回 true 。

• 第 16 至 28 行 ：呼叫 #isAfterSleepWindow() 方法，判斷是否滿足嘗試呼叫正常邏輯的間隔時間。當滿足，使用CAS 方式修改斷路器狀態( OPEN=>HALF_OPEN )，從而保證有且僅有一個執行緒能夠嘗試呼叫正常邏輯。

---

#isAfterSleepWindow() 方法，程式碼如下 ：

private boolean isAfterSleepWindow() {
    final long circuitOpenTime = circuitOpened.get();
    final long currentTime = System.currentTimeMillis();
    final long sleepWindowTime = properties.circuitBreakerSleepWindowInMilliseconds().get();
    return currentTime > circuitOpenTime + sleepWindowTime;
}

• 在當前時間超過斷路器開啟時間 HystrixCommandProperties.circuitBreakerSleepWindowInMilliseconds ( 預設值， 5000ms )，傳回 true 。

4.4 #markSuccess()

當嘗試呼叫正常邏輯成功時，呼叫 #markSuccess() 方法，關閉斷路器。程式碼如下 ：

  1: @Override
  2: public void markSuccess() {
  3:     if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
  4:         // 清空 Hystrix Metrics 對請求量統計 Observable 的**統計資訊**
  5:         //This thread wins the race to close the circuit - it resets the stream to start it over from 0
  6:         metrics.resetStream();
  7:         // 取消原有訂閱
  8:         Subscription previousSubscription = activeSubscription.get();
  9:         if (previousSubscription != null) {
 10:             previousSubscription.unsubscribe();
 11:         }
 12:         // 發起新的訂閱
 13:         Subscription newSubscription = subscribeToStream();
 14:         activeSubscription.set(newSubscription);
 15:         // 設定斷路器開啟時間為空
 16:         circuitOpened.set(-1L);
 17:     }
 18: }

• 第 3 行 ：使用 CAS 方式，修改斷路器狀態( HALF_OPEN=>CLOSED )。

• 第 6 行 ：清空 Hystrix Metrics 對請求量統計 Observable 的統計資訊。

• 第 8 至 14 行 ：取消原有訂閱，發起新的訂閱。

• 第 16 行 ：設定斷路器開啟時間為"空" 。

---

如下兩處呼叫了 #markNonSuccess() 方法 ：

• markEmits

• markOnCompleted

4.5 #markNonSuccess()

當嘗試呼叫正常邏輯失敗時，呼叫 #markNonSuccess() 方法，重新開啟斷路器。程式碼如下 ：

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

• 第 3 行 ：使用 CAS 方式，修改斷路器狀態( HALF_OPEN=>OPEN )。

• 第 5 行 ：設定設定斷路器開啟時間為當前時間。這樣， #attemptExecution() 過一段時間，可以再次嘗試執行正常邏輯。

---

如下兩處呼叫了 #markNonSuccess() 方法 ：

• handleFallback

• unsubscribeCommandCleanup

4.6 #allowRequest()

#allowRequest() 和 #attemptExecution() 方法，方法目的基本類似，差別在於當斷路器滿足嘗試關閉條件時，前者不會將斷路器不會修改狀態( CLOSE=>HALF-OPEN )，而後者會。點選 連結 檢視程式碼實現。

4.7 #isOpen()

#isOpen() 方法，比較簡單，點選 連結 檢視程式碼實現。

666. 彩蛋

呼呼，相對比較乾凈的一篇文章，滿足。

胖友，分享一波朋友圈可好！

    

芋艿

贊賞

長按二維碼向我轉賬

受蘋果公司新規定影響，微信 iOS 版的贊賞功能被關閉，可透過二維碼轉賬支援公眾號。

閱讀原文

即將開啟""小程式
取消
開啟