Netflix Hystrix斷路器原理分析

斷路器原理

斷路器在HystrixCommand和HystrixObservableCommand執行過程當中起到相當重要的做用。查看一下核心組件HystrixCircuitBreaker

package com.netflix.hystrix;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import com.netflix.hystrix.HystrixCommandMetrics.HealthCounts;
import rx.Subscriber;
import rx.Subscription;

public interface HystrixCircuitBreaker {

    boolean allowRequest();
    
    boolean isOpen();

    void markSuccess();

    void markNonSuccess();

    boolean attemptExecution();

    class Factory {
        // String is HystrixCommandKey.name() (we can't use HystrixCommandKey directly as we can't guarantee it implements hashcode/equals correctly)
        private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();
    }


    class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {
    }

    static class NoOpCircuitBreaker implements HystrixCircuitBreaker {
    }

}

下面先看一下該接口的抽象方法：

1. allowRequest(): 每一個Hystrix命令的請求都經過它判斷是否被執行(已經再也不使用,使用attemptExecution()方法進行判斷)
2. attemptExecution(): 每一個Hystrix命令的請求都經過它判斷是否被執行
3. isOpen(): 返回當前斷路器是否打開
4. markSuccess(): 用來關閉斷路器
5. markNonSuccess: 用來打開斷路器

下面看一下該接口中的類:

1. Factory: 維護了一個Hystrix命令和HystrixCircuitBreaker的關係的集合ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand。其中key經過HystrixCommandKey來定義，每個Hystrix命令都須要有一個Key來標識，同時根據這個Key能夠找到對應的斷路器實例。
2. NoOpCircuitBreaker: 一個啥都不作的斷路器，它容許全部請求經過，而且斷路器始終處於閉合狀態
3. HystrixCircuitBreakerImpl:斷路器的另外一個實現類。

HystrixCircuitBreakerImpl介紹

在該類中定義了斷路器的五個核心對象:

1. HystrixCommandProperties properties:斷路器對應實例的屬性集合對象
2. HystrixCommandMetrics metrics:用來讓HystrixCommand記錄各種度量指標的對象
3. AtomicReference<Status> status: 用來記錄斷路器的狀態，默認是關閉狀態
4. AtomicLong circuitOpened:斷路器打開的時間戳，默認-1，表示斷路器未打開
5. AtomicReference<Subscription> activeSubscription: 記錄HystrixCommand

isOpen方法介紹

@Override
        public boolean isOpen() {
            if (properties.circuitBreakerForceOpen().get()) {
                return true;
            }
            if (properties.circuitBreakerForceClosed().get()) {
                return false;
            }
            return circuitOpened.get() >= 0;
        }

用來判斷斷路器是否打開或關閉。主要步驟有：

1. 若是斷路器強制打開，返回true
2. 若是斷路器強制關閉，返回false
3. 判斷circuitOpened的值，若是大於等於0，返回true, 不然返回false

attemptExecution方法介紹

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

        @Override
        public boolean attemptExecution() {
            if (properties.circuitBreakerForceOpen().get()) {
                return false;
            }
            if (properties.circuitBreakerForceClosed().get()) {
                return true;
            }
            if (circuitOpened.get() == -1) {
                return true;
            } else {
                if (isAfterSleepWindow()) {
                    if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
                        //only the first request after sleep window should execute
                        return true;
                    } else {
                        return false;
                    }
                } else {
                    return false;
                }
            }
        }

該方法的主要邏輯有如下幾步:

1. 若是斷路器強制打開，返回false，不容許放過請求
2. 若是斷路器強制關閉，返回true，容許放過請求
3. 若是斷路器是關閉狀態，返回true，容許放過請求
4. 判斷當前時間是否超過斷路器打開的時間加上滑動窗口的時間，若是沒有超過，返回false，不容許放過請求
5. 若是沒有超過，若是斷路器是打開狀態，而且設置斷路器狀態爲半開狀態成功時，返回true，容許放過請求
6. 若是失敗，則返回false，不容許放過請求

markSuccess方法

@Override
        public void markSuccess() {
            if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
                //This thread wins the race to close the circuit - it resets the stream to start it over from 0
                metrics.resetStream();
                Subscription previousSubscription = activeSubscription.get();
                if (previousSubscription != null) {
                    previousSubscription.unsubscribe();
                }
                Subscription newSubscription = subscribeToStream();
                activeSubscription.set(newSubscription);
                circuitOpened.set(-1L);
            }
        }

該方法主要用來關閉斷路器，主要邏輯有如下幾步：

1. 若是斷路器狀態是半開而且成功設置爲關閉狀態時，執行如下步驟。
2. 重置度量指標對象
3. 取消以前的訂閱，發起新的訂閱
4. 設置斷路器的打開時間爲-1

代碼地址

spring-cloud-example