Hystrix

簡要說明

在大中型分布式系統(tǒng)中，通常系統(tǒng)很多依賴。在高并發(fā)訪問下,這些依賴的穩(wěn)定性與否對系統(tǒng)的影響非常大,但是依賴有很多不可控問題:如網(wǎng)絡(luò)連接緩慢，資源繁忙，暫時不可用，服務(wù)脫機等。在復雜的分布式架構(gòu)的應(yīng)用程序有很多的依賴，都會不可避免地在某些時候失敗。高并發(fā)的依賴失敗時如果沒有隔離措施，當前應(yīng)用服務(wù)就有被拖垮的風險。一般來說，隨著服務(wù)依賴數(shù)量的變多，服務(wù)不穩(wěn)定的概率會成指數(shù)性提高。

解決的問題

一個依賴30個SOA服務(wù)的系統(tǒng),每個服務(wù)99.99%可用。
99.99%的30次方 ≈ 99.7%，
0.3% 意味著一億次請求 會有 3,000,00次失敗， 換算成時間大約每月有2個小時服務(wù)不穩(wěn)定。 解決這個問題的方案是對依賴進行隔離。Hystrix就是處理依賴隔離的框架,同時也是可以幫我們做依賴服務(wù)的治理和監(jiān)控。

前期準備

rxjava

Hystrix引入rxjava 1
四個基本概念：Observable (可觀察者，即被觀察者)、 Observer (觀察者)、 subscribe (訂閱)、事件。Observable 和 Observer 通過 subscribe() 方法實現(xiàn)訂閱關(guān)系，從而 Observable 可以在需要的時候發(fā)出事件來通知 Observer。
BlockingObservable<T> ，一個阻塞的Observable 繼承普通的Observable類，增加了一些可用于阻塞Observable發(fā)射的數(shù)據(jù)的操作符。BlockingObservable已經(jīng)在Rxjava2中去掉了，在Rxjava2中已經(jīng)集成到了Observable

線程池、Future

略

流程圖

flow chart

構(gòu)建一個HystrixCommand 或 HystrixObservableCommand

  HystrixCommand   
    R run()
    R getFallback()
  HystrixObservableCommand    
    Observable<R> construct()
    resumeWithFallback()

執(zhí)行Command

    K value = command.execute(); -> queue().get()
    Future<K> fValue  = command.queue(); -> toObservable().toBlocking().toFuture();( Observable -> BlockingObservable -> Future(delegate) -> Future)
    Observable<K> ohValue = command.observe();         //hot observable    
    Observable<K> ocValue = command.toObservable();    //cold observable

    public Observable<R> observe() {
        // us a ReplaySubject to buffer the eagerly subscribed-to Observable
        ReplaySubject<R> subject = ReplaySubject.create();
        // eagerly kick off subscription
        final Subscription sourceSubscription = toObservable().subscribe(subject);
        // return the subject that can be subscribed to later while the execution has already started
        return subject.doOnUnsubscribe(new Action0() {
            @Override
            public void call() {
                sourceSubscription.unsubscribe();
            }
        });
    }

command chart

是否緩存

重寫getCacheKey()，用來構(gòu)造cache key。
HystrixRequestContext.initializeContext()和context.shutdown()構(gòu)建context。

    public class HelloWorld extends HystrixCommand<Boolean> {
        @Override
        protected String getCacheKey() {
            return ?;
        }
        
        public static void main(String[] args) throws Exception {
            HystrixRequestContext context = HystrixRequestContext.initializeContext();
            try {
                ...
            }finally {
                context.shutdown();
            }
        }
    }

Circuit是否打開

檢測circuit-breaker是否打開，打開則直接進入fallback。否則進入下一步。
HystrixCommandProperties設(shè)置circuitBreaker

circuitBreaker.enabled 
設(shè)置斷路器是否生效。

circuitBreaker.requestVolumeThreshold
設(shè)置在一個滾動窗口中，打開斷路器的最少請求數(shù)。

circuitBreaker.sleepWindowInMilliseconds
設(shè)置在斷路器被打開，拒絕請求到再次嘗試請求的時間間隔。

circuitBreaker.errorThresholdPercentage
設(shè)置打開斷路器并啟動回退邏輯的錯誤比率。（這個參數(shù)的效果受到circuitBreaker.requestVolumeThreshold和滾動時間窗口的時間長度影響）

circuitBreaker.forceOpen
強制斷路器進入打開狀態(tài)

circuitBreaker.forceClosed
強制斷路器進入關(guān)閉狀態(tài)

線程池、隊列、信號是否占滿

線程池、隊列、信號是否占滿的時候，將直接進入fallback

線程池配置

coreSize 
心線程池的大小。

maximumSize
設(shè)置線程池最大值。

maxQueueSize
設(shè)置BlockingQueue最大的隊列值。

queueSizeRejectionThreshold
設(shè)置隊列拒絕的閾值

keepAliveTimeMinutes
設(shè)置存活時間，單位分鐘。

allowMaximumSizeToDivergeFromCoreSize
該屬性允許maximumSize起作用。

metrics.rollingStats.timeInMilliseconds
設(shè)置統(tǒng)計的滾動窗口的時間段大小。

metrics.rollingStats.numBuckets
設(shè)置滾動的統(tǒng)計窗口被分成的bucket的數(shù)目。

Circuit Breaker chart

run()或者construct()

執(zhí)行相應(yīng)的run()或者construct()的方法。

計算Circuit Health

通過計算successes, failures, rejections, and timeouts，確定是否打開circuitBreaker。

getFallback()或者resumeWithFallback()

是否進入fallback

    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);
                }
            }
        };

折疊器

通過繼承HystrixCollapser，實現(xiàn)多個請求折疊成單個請求。

collapser chart

返回成功響應(yīng)