熔斷器 Hystrix 原始碼解析 —— 命令執行（二）之執行隔離策略

阿新 • • 發佈：2022-05-02

本文主要基於 Hystrix 1.5.X 版本

1. 概述
2. HystrixThreadPoolProperties
3. HystrixThreadPoolKey
4. HystrixConcurrencyStrategy
5. HystrixThreadPool
6. HystrixScheduler

1. 概述

本文主要分享 Hystrix 命令執行（二）之執行隔離策略。

建議：對 RxJava 已經有一定的瞭解的基礎上閱讀本文。

Hystrix 提供兩種執行隔離策略( ExecutionIsolationStrategy ) ：

SEMAPHORE ：訊號量，命令在呼叫執行緒

執行。在《Hystrix 原始碼解析 —— 命令執行（一）之正常執行邏輯》「3. TryableSemaphore」已經詳細解析。
THREAD ：執行緒池，命令在執行緒池執行。在《Hystrix 原始碼解析 —— 命令執行（一）之正常執行邏輯》「5. #executeCommandWithSpecifiedIsolation(...)」的 #executeCommandWithSpecifiedIsolation(...) 方法中，呼叫 Observable#subscribeOn(Scheduler) 方法，指定在 RxJava Scheduler 執行。
- 如果你暫時不瞭解 Scheduler ，可以閱讀《RxJava 原始碼解析 —— Scheduler》。
- 如果你暫時不瞭解 Observable#subscribeOn(Scheduler) ，可以閱讀《RxJava 原始碼解析 —— Observable#subscribeOn(Scheduler)》。

兩種方式的優缺點比較，推薦閱讀《【翻譯】Hystrix文件-實現原理》「依賴隔離」。

2. HystrixThreadPoolProperties

com.netflix.hystrix.HystrixThreadPoolProperties ，Hystrix 執行緒池屬性配置抽象類，點選連結檢視，已新增中文註釋說明。

com.netflix.hystrix.strategy.properties.HystrixPropertiesThreadPoolDefault

，Hystrix 執行緒池配置實現類，點選連結檢視。實際上沒什麼內容，官方如是說：

Default implementation of {@link HystrixThreadPoolProperties} using Archaius (https://github.com/Netflix/archaius)

3. HystrixThreadPoolKey

com.netflix.hystrix.HystrixThreadPoolKey ，Hystrix 執行緒池標識介面。

FROM HystrixThreadPoolKey 介面註釋 A key to represent a {@link HystrixThreadPool} for monitoring, metrics publishing, caching and other such uses. This interface is intended to work natively with Enums so that implementing code can be an enum that implements this interface.

直白的說，希望通過相同的 name ( 標識 ) 獲得同 HystrixThreadPoolKey 物件。通過在內部維持一個 name 與 HystrixThreadPoolKey 物件的對映，以達到列舉的效果。

HystrixThreadPoolKey 程式碼如下：

1: public interface HystrixThreadPoolKey extends HystrixKey {
  2:     class Factory {
  3:         private Factory() {
  4:         }
  5: 
  6:         // used to intern instances so we don't keep re-creating them millions of times for the same key
  7:         private static final InternMap<String, HystrixThreadPoolKey> intern
  8:                 = new InternMap<String, HystrixThreadPoolKey>(
  9:                 new InternMap.ValueConstructor<String, HystrixThreadPoolKey>() {
 10:                     @Override
 11:                     public HystrixThreadPoolKey create(String key) {
 12:                         return new HystrixThreadPoolKeyDefault(key);
 13:                     }
 14:                 });
 15: 
 16:         public static HystrixThreadPoolKey asKey(String name) {
 17:            return intern.interned(name);
 18:         }
 19: 
 20:         private static class HystrixThreadPoolKeyDefault extends HystrixKeyDefault implements HystrixThreadPoolKey {
 21:             public HystrixThreadPoolKeyDefault(String name) {
 22:                 super(name);
 23:             }
 24:         }
 25: 
 26:         /* package-private */ static int getThreadPoolCount() {
 27:             return intern.size();
 28:         }
 29:     }
 30: }

HystrixThreadPoolKey 實現 com.netflix.hystrix.HystrixKey 介面，點選連結檢視。該介面定義的 #name()方法，即是上文我們所說的標識( Key )。
intern 屬性， name 與 HystrixThreadPoolKey 物件的對映，以達到列舉的效果。
- com.netflix.hystrix.util.InternMap ，點選連結檢視帶中文註釋的程式碼。
#asKey(name) 方法，從 intern 獲得 HystrixThreadPoolKey 物件。
#getThreadPoolCount() 方法，獲得 HystrixThreadPoolKey 數量。

在 AbstractCommand 構造方法裡，初始化命令的 threadPoolKey 屬性，程式碼如下：

protected final HystrixThreadPoolKey threadPoolKey;

protected AbstractCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,
       HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,
       HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,
       HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook) {

   // ... 省略無關程式碼

   this.commandGroup = initGroupKey(group);
   this.commandKey = initCommandKey(key, getClass());
   this.properties = initCommandProperties(this.commandKey, propertiesStrategy, commandPropertiesDefaults);
   // 初始化 threadPoolKey
   this.threadPoolKey = initThreadPoolKey(threadPoolKey, this.commandGroup, this.properties.executionIsolationThreadPoolKeyOverride().get());

}

呼叫 #initThreadPoolKey(...) 方法，建立最終的 threadPoolKey 屬性。程式碼如下：
- 優先順序： threadPoolKeyOverride > threadPoolKey > groupKey
1. private static HystrixThreadPoolKey initThreadPoolKey(HystrixThreadPoolKey threadPoolKey, HystrixCommandGroupKey groupKey, String threadPoolKeyOverride) {
2. if (threadPoolKeyOverride == null) {
3. // we don't have a property overriding the value so use either HystrixThreadPoolKey or HystrixCommandGroup
4. if (threadPoolKey == null) {
5. /* use HystrixCommandGroup if HystrixThreadPoolKey is null */
6. return HystrixThreadPoolKey.Factory.asKey(groupKey.name());
7. } else {
8. return threadPoolKey;
9. }
10. } else { // threadPoolKeyOverride 可覆蓋屬性
11. // we have a property defining the thread-pool so use it instead
12. return HystrixThreadPoolKey.Factory.asKey(threadPoolKeyOverride);
13. }
14. }

4. HystrixConcurrencyStrategy

com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy ，Hystrix 併發策略抽象類。

HystrixConcurrencyStrategy#getThreadPool(...) 方法，程式碼如下：

1: public ThreadPoolExecutor getThreadPool(final HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties threadPoolProperties) {
  2:     final ThreadFactory threadFactory = getThreadFactory(threadPoolKey);
  3: 
  4:     final boolean allowMaximumSizeToDivergeFromCoreSize = threadPoolProperties.getAllowMaximumSizeToDivergeFromCoreSize().get();
  5:     final int dynamicCoreSize = threadPoolProperties.coreSize().get();
  6:     final int keepAliveTime = threadPoolProperties.keepAliveTimeMinutes().get();
  7:     final int maxQueueSize = threadPoolProperties.maxQueueSize().get();
  8: 
  9:     final BlockingQueue<Runnable> workQueue = getBlockingQueue(maxQueueSize);
 10: 
 11:     if (allowMaximumSizeToDivergeFromCoreSize) {
 12:         final int dynamicMaximumSize = threadPoolProperties.maximumSize().get();
 13:         if (dynamicCoreSize > dynamicMaximumSize) {
 14:             logger.error("Hystrix ThreadPool configuration at startup for : " + threadPoolKey.name() + " is trying to set coreSize = " +
 15:                     dynamicCoreSize + " and maximumSize = " + dynamicMaximumSize + ".  Maximum size will be set to " +
 16:                     dynamicCoreSize + ", the coreSize value, since it must be equal to or greater than the coreSize value");
 17:             return new ThreadPoolExecutor(dynamicCoreSize, dynamicCoreSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);
 18:         } else {
 19:             return new ThreadPoolExecutor(dynamicCoreSize, dynamicMaximumSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);
 20:         }
 21:     } else {
 22:         return new ThreadPoolExecutor(dynamicCoreSize, dynamicCoreSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);
 23:     }
 24: }

第 2 行：呼叫 #getThreadFactory(...) 方法，獲得 ThreadFactory 。點選連結檢視方法程式碼。
- PlatformSpecific#getAppEngineThreadFactory() 方法，無需細看，適用於 Google App Engine 場景。
第 4 至 7 行：「2. HystrixThreadPoolProperties」有詳細解析。
第 9 行：呼叫 #getBlockingQueue() 方法，獲得執行緒池的阻塞佇列。點選連結檢視方法程式碼。
- 《Java阻塞佇列ArrayBlockingQueue和LinkedBlockingQueue實現原理分析》
- 《Java併發包中的同步佇列SynchronousQueue實現原理》
- 當 maxQueueSize<=0 時( 預設值： -1 ) 時，使用 SynchronousQueue 。超過執行緒池的 maximumPoolSize時，提交任務被拒絕。
- 當 SynchronousQueue>0 時，使用 LinkedBlockingQueue 。超過執行緒池的 maximumPoolSize 時，任務被拒絕。超過執行緒池的 maximumPoolSize + 執行緒池佇列的 maxQueueSize 時，提交任務被阻塞等待。
推薦：《聊聊併發（三）——JAVA執行緒池的分析和使用》
推薦：《聊聊併發（七）——Java中的阻塞佇列》
第 11 至 23 行：建立 ThreadPoolExecutor 。看起來程式碼比較多，根據 allowMaximumSizeToDivergeFromCoreSize的情況，計算執行緒池的 maximumPoolSize 屬性。計算的方式和 HystrixThreadPoolProperties#actualMaximumSize() 方法是一致的。

com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategyDefault ，Hystrix 併發策略實現類。程式碼如下( 基本沒做啥 ) ：

public class HystrixConcurrencyStrategyDefault extends HystrixConcurrencyStrategy {

    /**
     * 單例
     */
    private static HystrixConcurrencyStrategyDefault INSTANCE = new HystrixConcurrencyStrategyDefault();

    public static HystrixConcurrencyStrategy getInstance() {
        return INSTANCE;
    }

    private HystrixConcurrencyStrategyDefault() {
    }

}

在 AbstractCommand 構造方法裡，初始化命令的 threadPoolKey 屬性，程式碼如下：

protected final HystrixConcurrencyStrategy concurrencyStrategy;

protected AbstractCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,
       HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,
       HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,
       HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook) {

    // ... 省略無關程式碼

    // 初始化 併發策略
    this.concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();

}

HystrixPlugins ，Hystrix 外掛體系，https://github.com/Netflix/Hystrix/wiki/Plugins 有詳細解析。
呼叫 HystrixPlugins#getConcurrencyStrategy() 獲得 HystrixConcurrencyStrategy 物件。預設情況下，使用 HystrixConcurrencyStrategyDefault 。當然你也可以參考 Hystrix 外掛體系，實現自定義的 HystrixConcurrencyStrategy 實現，以達到覆寫 #getThreadPool()， #getBlockingQueue() 等方法。點選連結檢視該方法程式碼。

5. HystrixThreadPool

com.netflix.hystrix.HystrixThreadPool ，Hystrix 執行緒池介面。當 Hystrix 命令使用 THREAD 執行隔離策略時， HystrixCommand#run() 方法在執行緒池執行。點選連結檢視。HystrixThreadPool 定義介面如下：

#getExecutor() ：獲得 ExecutorService 。
#getScheduler() / #getScheduler(Func0<Boolean>) ：獲得 RxJava Scheduler 。
#isQueueSpaceAvailable() ：執行緒池佇列是否有空餘。
#markThreadExecution() / #markThreadCompletion() / #markThreadRejection() ：TODO 【2002】【metrics】

5.1 HystrixThreadPoolDefault

com.netflix.hystrix.HystrixThreadPool.HystrixThreadPoolDefault ，Hystrix 執行緒池實現類。

構造方法，程式碼如下：

1: private final HystrixThreadPoolProperties properties;
  2: private final BlockingQueue<Runnable> queue;
  3: private final ThreadPoolExecutor threadPool;
  4: private final HystrixThreadPoolMetrics metrics;
  5: private final int queueSize;
  6: 
  7: public HystrixThreadPoolDefault(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter propertiesDefaults) {
  8:     // 初始化 HystrixThreadPoolProperties
  9:     this.properties = HystrixPropertiesFactory.getThreadPoolProperties(threadPoolKey, propertiesDefaults);
 10:     // 獲得 HystrixConcurrencyStrategy
 11:     HystrixConcurrencyStrategy concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
 12:     // 佇列大小
 13:     this.queueSize = properties.maxQueueSize().get();
 14: 
 15:     // TODO 【2002】【metrics】
 16:     this.metrics = HystrixThreadPoolMetrics.getInstance(threadPoolKey,
 17:             concurrencyStrategy.getThreadPool(threadPoolKey, properties), // 初始化 ThreadPoolExecutor
 18:             properties);
 19: 
 20:     // 獲得 ThreadPoolExecutor
 21:     this.threadPool = this.metrics.getThreadPool();
 22:     this.queue = this.threadPool.getQueue(); // 佇列
 23: 
 24:     // TODO 【2002】【metrics】
 25:     /* strategy: HystrixMetricsPublisherThreadPool */
 26:     HystrixMetricsPublisherFactory.createOrRetrievePublisherForThreadPool(threadPoolKey, this.metrics, this.properties);
 27: }

第 9 行：初始化 HystrixThreadPoolProperties 。
第 11 行：初始化 HystrixConcurrencyStrategy 。
第 13 行：初始化 queueSize 。
第 16 至 18 行：TODO 【2002】【metrics】
- 第 17 行：呼叫 HystrixConcurrencyStrategy#getThreadPool(...) 方法，初始化 ThreadPoolExecutor 。
第 21 行：獲得 ThreadPoolExecutor 。
第 22 行：獲得 ThreadPoolExecutor 的佇列。
第 26 行：TODO 【2002】【metrics】

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

@Override
public ThreadPoolExecutor getExecutor() {
    touchConfig();
    return threadPool;
}

呼叫 #touchConfig() 方法，動態調整 threadPool 的 coreSize / maximumSize / keepAliveTime 引數。點選連結檢視該方法。

#getScheduler() / #getScheduler(Func0<Boolean>) 方法，程式碼如下：

@Override
public Scheduler getScheduler() {
    //by default, interrupt underlying threads on timeout
    return getScheduler(new Func0<Boolean>() {
        @Override
        public Boolean call() {
            return true;
        }
    });
}

@Override
public Scheduler getScheduler(Func0<Boolean> shouldInterruptThread) {
    touchConfig();
    return new HystrixContextScheduler(HystrixPlugins.getInstance().getConcurrencyStrategy(), this, shouldInterruptThread);
}

HystrixContextScheduler 和 shouldInterruptThread 都在「6. HystrixContextScheduler」詳細解析。

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

@Override
public boolean isQueueSpaceAvailable() {
    if (queueSize <= 0) {
        // we don't have a queue so we won't look for space but instead
        // let the thread-pool reject or not
        return true;
    } else {
        return threadPool.getQueue().size() < properties.queueSizeRejectionThreshold().get();
    }
}

由於執行緒池的佇列大小不能動態調整，該方法的實現通過 HystrixThreadPoolProperties.queueSizeRejectionThreshold 屬性控制。
注意 queueSize 屬性，決定了執行緒池的佇列型別。
- queueSize<=0 時， #isQueueSpaceAvailable() 都返回 true 的原因是，執行緒池使用 SynchronousQueue 作為佇列，不支援新任務排隊，任務超過執行緒池的 maximumPoolSize 時，新任務被拒絕。
- queueSize>0 時， #isQueueSpaceAvailable() 根據情況 true/ false 的原因是，執行緒池使用 LinkedBlockingQueue 作為佇列，支援一定數量的阻塞排隊，但是這個數量無法調整。通過 #isQueueSpaceAvailable() 方法的判斷，動態調整。另外，初始配置的 queueSize 要相對大，否則即使 queueSizeRejectionThreshold 配置的大於 queueSize ，實際提交任務到執行緒池，也會被拒絕。

5.2 Factory

com.netflix.hystrix.HystrixThreadPool.Factory ，HystrixThreadPool 工廠類，不僅限於 HystrixThreadPool 的建立，也提供了 HystrixThreadPool 的管理( HystrixThreadPool 的容器 )。

threadPools 屬性，維護建立的 HystrixThreadPool 對應的對映，程式碼如下：

final static ConcurrentHashMap<String, HystrixThreadPool> threadPools = new ConcurrentHashMap<String, HystrixThreadPool>();

Key 為 HystrixThreadPoolKey#name() ，每個 HystrixThreadPoolKey 對應一個 HystrixThreadPool 物件。

#getInstance(...) 方法，獲得 HystrixThreadPool 物件，程式碼如下：

/* package */static HystrixThreadPool getInstance(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter propertiesBuilder) {
    // get the key to use instead of using the object itself so that if people forget to implement equals/hashcode things will still work
    String key = threadPoolKey.name();

    // this should find it for all but the first time
    HystrixThreadPool previouslyCached = threadPools.get(key);
    if (previouslyCached != null) {
         return previouslyCached;
     }

     // if we get here this is the first time so we need to initialize
     synchronized (HystrixThreadPool.class) {
        if (!threadPools.containsKey(key)) {
            threadPools.put(key, new HystrixThreadPoolDefault(threadPoolKey, propertiesBuilder));
       }
     }
     return threadPools.get(key);
}

根據 threadPoolKey 先從 threadPool 獲取已建立的 HystrixThreadPool ；獲取不到，建立對應的 HystrixThreadPool 返回，並新增到 threadPool 。

#shutdown() / #shutdown(timeout, unit) 方法，比較易懂，點選連結檢視。

5.3 初始化

在 AbstractCommand 構造方法裡，初始化命令的 threadPool 屬性，程式碼如下：

protected final HystrixThreadPool threadPool;

protected AbstractCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,
       HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,
       HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,
       HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook) {

    // ... 省略其他程式碼

    // 初始化 threadPoolKey
    this.threadPoolKey = initThreadPoolKey(threadPoolKey, this.commandGroup, this.properties.executionIsolationThreadPoolKeyOverride().get());
    // 初始化 threadPool
    this.threadPool = initThreadPool(threadPool, this.threadPoolKey, threadPoolPropertiesDefaults);

}

呼叫 #initThreadPool(...) 方法，獲得 HystrixThreadPool ，點選連結檢視。

6. HystrixScheduler

Hystrix 實現了自定義的 RxJava Scheduler ，整體類圖如下：

HystrixContextScheduler ( 實現 RxJava Scheduler 抽象類 )，內嵌型別為 ThreadPoolScheduler ( 實現 RxJava Scheduler抽象類 )的 actualScheduler 屬性。
HystrixContextWorker ( 實現 RxJava Worker 抽象類 )，內嵌型別為 ThreadPoolWorker ( 實現 RxJava Worker 抽象類 )的 worker 屬性。

6.1 HystrixContextScheduler

構造方法，程式碼如下：

public class HystrixContextScheduler extends Scheduler {

    private final HystrixConcurrencyStrategy concurrencyStrategy;
    private final Scheduler actualScheduler;
    private final HystrixThreadPool threadPool;

    // ... 省略無關程式碼

    public HystrixContextScheduler(HystrixConcurrencyStrategy concurrencyStrategy, HystrixThreadPool threadPool, Func0<Boolean> shouldInterruptThread) {
        this.concurrencyStrategy = concurrencyStrategy;
        this.threadPool = threadPool;
        this.actualScheduler = new ThreadPoolScheduler(threadPool, shouldInterruptThread);
    }
}

actualScheduler 屬性，型別為 ThreadPoolScheduler 。

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

@Override
public Worker createWorker() {
    return new HystrixContextSchedulerWorker(actualScheduler.createWorker());
}

使用 actualScheduler 建立 ThreadPoolWorker ，傳參給 HystrixContextSchedulerWorker 。

6.2 HystrixContextSchedulerWorker

構造方法，程式碼如下：

private class HystrixContextSchedulerWorker extends Worker {

    private final Worker worker;

    // ... 省略無關程式碼

    private HystrixContextSchedulerWorker(Worker actualWorker) {
        this.worker = actualWorker;
    }

}

worker 屬性，型別為 ThreadPoolWorker 。

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

@Override
public Subscription schedule(Action0 action) {
    if (threadPool != null) {
        if (!threadPool.isQueueSpaceAvailable()) {
            throw new RejectedExecutionException("Rejected command because thread-pool queueSize is at rejection threshold.");
        }
    }
    return worker.schedule(new HystrixContexSchedulerAction(concurrencyStrategy, action));
}

呼叫 ThreadPool#isQueueSpaceAvailable() 方法，判斷執行緒池佇列是否有空餘。這個就是 HystrixContextScheduler 的實際用途。

#unsubscribe() / #isUnsubscribed() 方法，使用 worker 判斷，點選連結檢視。

6.3 ThreadPoolScheduler

ThreadPoolScheduler 比較簡單，點選連結檢視。

6.4 ThreadPoolWorker

構造方法，程式碼如下：

private static class ThreadPoolWorker extends Worker {

    private final HystrixThreadPool threadPool;
    private final CompositeSubscription subscription = new CompositeSubscription();
    private final Func0<Boolean> shouldInterruptThread;

    // ... 省略無關程式碼

    public ThreadPoolWorker(HystrixThreadPool threadPool, Func0<Boolean> shouldInterruptThread) {
        this.threadPool = threadPool;
        this.shouldInterruptThread = shouldInterruptThread;
    }
}

subscription 屬性，訂閱資訊。

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

1: @Override
  2: public Subscription schedule(final Action0 action) {
  3:     // 未訂閱，返回
  4:     if (subscription.isUnsubscribed()) {
  5:         // don't schedule, we are unsubscribed
  6:         return Subscriptions.unsubscribed();
  7:     }
  8: 
  9:     // 建立 ScheduledAction
 10:     // This is internal RxJava API but it is too useful.
 11:     ScheduledAction sa = new ScheduledAction(action);
 12: 
 13:     // 新增到 訂閱
 14:     subscription.add(sa);
 15:     sa.addParent(subscription);
 16: 
 17:     // 提交 任務
 18:     ThreadPoolExecutor executor = (ThreadPoolExecutor) threadPool.getExecutor();
 19:     FutureTask<?> f = (FutureTask<?>) executor.submit(sa);
 20:     sa.add(new FutureCompleterWithConfigurableInterrupt(f, shouldInterruptThread, executor));
 21: 
 22:     return sa;
 23: }

第 4 至 7 行：未訂閱，返回。
第 11 行：建立 ScheduledAction 。在 TODO 【2013】【ScheduledAction】詳細解析。
第 14 至 15 行：新增到訂閱( subscription )。
第 18 至 20 行：使用 threadPool ，提交任務，並建立 FutureCompleterWithConfigurableInterrupt 新增到訂閱( sa)。
第 22 行：返回訂閱( sa )。整體訂閱關係如下：

#unsubscribe() / #isUnsubscribed() 方法，使用 subscription 判斷，點選連結檢視。

6.5 FutureCompleterWithConfigurableInterrupt

com.netflix.hystrix.strategy.concurrency.HystrixContextScheduler.FutureCompleterWithConfigurableInterrupt ，實現類似 rx.internal.schedulers.ScheduledAction.FutureCompleter ，在它的基礎上，支援配置 FutureTask#cancel(Boolean) 是否可打斷執行( mayInterruptIfRunning )。

構造方法，程式碼如下：

private static class FutureCompleterWithConfigurableInterrupt implements Subscription {
    private final FutureTask<?> f;
    private final Func0<Boolean> shouldInterruptThread;
    private final ThreadPoolExecutor executor;

    // ... 省略無關程式碼

    private FutureCompleterWithConfigurableInterrupt(FutureTask<?> f, Func0<Boolean> shouldInterruptThread, ThreadPoolExecutor executor) {
        this.f = f;
        this.shouldInterruptThread = shouldInterruptThread;
        this.executor = executor;
    }
}

當命令執行超時，或是主動取消命令執行時，呼叫 #unsubscribe() 方法，取消執行。 當命令執行超時，或是主動取消命令執行時，呼叫 #unsubscribe() 方法，取消執行。 當命令執行超時，或是主動取消命令執行時，呼叫 #unsubscribe() 方法，取消執行。

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

@Override
public void unsubscribe() {
    // 從 執行緒池 移除 任務
    executor.remove(f);
    // 根據 shouldInterruptThread 配置，是否強制取消
    if (shouldInterruptThread.call()) {
        f.cancel(true);
    } else {
        f.cancel(false);
    }
}

根據 shouldInterruptThread 方法，判斷是否強制取消。
shouldInterruptThread 對應的方法，實現程式碼如下：

subscribeOn(threadPool.getScheduler(new Func0<Boolean>() {
    @Override
    public Boolean call() {
        return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;
    }
}));

當 executionIsolationThreadInterruptOnTimeout=true 時，命令可執行超時。當命令可執行超時時，強制取消。
當使用 HystrixCommand.queue() 返回的 Future ，可以使用 Future#cancel(Boolean) 取消命令執行。從 shouldInterruptThread 對應的方法可以看到，如果此時不滿足命令執行超時的條件，命令執行取消的方式是非強制的。此時當 executionIsolationThreadInterruptOnFutureCancel=true 時，並且呼叫 Future#cancel(Boolean) 傳遞 mayInterruptIfRunning=true ，強制取消命令執行。
- 模擬測試用例： CommandHelloWorld#testAsynchronous3()
- HystrixCommand#queue() ：點選連結檢視 Future#cancel(Boolean) 方法。

熔斷器 Hystrix 原始碼解析 —— 命令執行（二）之執行隔離策略

1. 概述

2. HystrixThreadPoolProperties

3. HystrixThreadPoolKey

4. HystrixConcurrencyStrategy

5. HystrixThreadPool

5.1 HystrixThreadPoolDefault

5.2 Factory

5.3 初始化

6. HystrixScheduler

6.1 HystrixContextScheduler

6.2 HystrixContextSchedulerWorker

6.3 ThreadPoolScheduler

6.4 ThreadPoolWorker

6.5 FutureCompleterWithConfigurableInterrupt

熔斷器 Hystrix 原始碼解析 —— 命令執行（二）之執行隔離策略

熔斷器 Hystrix 原始碼解析 —— 命令執行（一）之正常執行邏輯

熔斷器 Hystrix 原始碼解析 —— 命令合併執行

Spring IoC 原始碼分析 (基於註解) （二）之包掃描

資料庫分庫分表中介軟體 Sharding-JDBC 原始碼分析 —— SQL 路由（二）之分庫分表路由

熔斷器 Hystrix 原始碼解析 —— 執行結果快取

mybatis 啟動流程原始碼分析（二）之 Configuration-Properties解析

Sharding-JDBC 原始碼分析 —— SQL 解析（二）之SQL解析

熔斷器 Hystrix 原始碼解析 —— 斷路器 HystrixCircuitBreaker

RocketMq原始碼分析篇Producer（二）

Spring IoC 原始碼分析 (基於註解)（三）之 Bean的解析與註冊

myBatis原始碼解析-日誌篇（1）

myBatis原始碼解析-快取篇（2）

myBatis原始碼解析-反射篇（4）

JVM基礎（二）：執行時資料區

Serilog 原始碼解析——Demo 實現（下）

JAVA多執行緒（二）：執行緒上下文切換、死鎖及守護執行緒

Docker系列（二）之Docker常用命令

mybaits原始碼分析--反射模組（二）

Linux常用命令總結（二）

熔斷器 Hystrix 原始碼解析 —— 命令執行（二）之執行隔離策略

1. 概述

2. HystrixThreadPoolProperties

3. HystrixThreadPoolKey

4. HystrixConcurrencyStrategy

5. HystrixThreadPool

5.1 HystrixThreadPoolDefault

5.2 Factory

5.3 初始化

6. HystrixScheduler

6.1 HystrixContextScheduler

6.2 HystrixContextSchedulerWorker

6.3 ThreadPoolScheduler

6.4 ThreadPoolWorker

6.5 FutureCompleterWithConfigurableInterrupt

相關推薦