碼上快樂

相關內容    簡體    繁體

Sentinel 熔斷等指標如何統計以及如何判斷熔斷點

本文轉載自   查看原文   2020-07-29 14:31   845    javaEE/ tools

Sentinel 使用

同時發布：http://fantasylion.github.io/java/2020-07-29-Sentinel-Source-code-analysis/

 

在分析源碼之前首先看下，Sentinel 如何使用

建立規則
1
2
3
4
5
6
7
8
9
10
11
12
13
 
// 建立規則
List<DegradeRule> rule = new ArrayList<DegradeRule>();
DegradeRule ruleRatio = new DegradeRule();
ruleRatio.setResource("sourceTest");
ruleRatio.setCount(100);
ruleRatio.setGrade(1);
ruleRatio.setTimeWindow(60)
ruleRatio.setMinRequestAmount(2);
ruleRatio.setRtSlowRequestAmount(2);
rules.add(ruleRatio);

// 加載規則
DegradeRuleManager.loadRules(rules);
使用規則
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
 
Entry entry = null;
try {
 entry = SphU.entry( "sourceTest" )
 print("Do something.");
} catch( DegradeException degradeException ) {
 logger.error("觸發熔斷,熔斷器：{}", JSON.toJSONString(degradeException.rule) )
 throw new DegradeException("觸發熔斷"+degradeException.rule.resource, degradeException)
} catch (Exception e) {
 Tracer.trace(e)
 logger.error("有異常")
 throw e
} finally {
 if (entry != null) {
 // 退出 Entry 並統計
 entry.exit()
 }
}

從上面的代碼中大致可以看出，sentinel 通過 SphU.entry 驗證規則並開始統計，如果其中某條規則不通過將會拋出對應的異常， 通過 entry.exit() 結束統計。

下面進入到源碼中分析具體的實現原理
CtSph類圖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
public static final Sph sph = new CtSph();

public static Entry entry(String name) throws BlockException {
 return Env.sph.entry(name, EntryType.OUT, 1, OBJECTS0); // @1 -> @2
}

// @2
// Env.sph.entry
public Entry entry(String name, EntryType type, int count, Object... args) throws BlockException {
 // 創建一個資源名包裝類
 StringResourceWrapper resource = new StringResourceWrapper(name, type);
 return entry(resource, count, args); // @3 -> @4
}

// @4
public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
 return entryWithPriority(resourceWrapper, count, false, args); // @5 -> @6
}

// @6
private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
 throws BlockException {
 // 從線程變量中獲取當前上下文
 Context context = ContextUtil.getContext();
 // ... 省略部分代碼
 if (context == null) {
 // Using default context.
 // 如果沒有上下文，創建一個默認的上下文和一個EntranceNode
 context = InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME);
 }

 // 如果全局開關關閉，不需要檢查規則和統計
 // Global switch is close, no rule checking will do.
 if (!Constants.ON) {
 return new CtEntry(resourceWrapper, null, context);
 }

 // 找到所有的處理責任鏈【責任鏈模式】
 ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);

 /*
 * 說明責任鏈數量已經超出最大允許數量，后面將沒有規則會被檢查
 * Means amount of resources (slot chain) exceeds {@link Constants.MAX_SLOT_CHAIN_SIZE},
 * so no rule checking will be done.
 */
 if (chain == null) {
 return new CtEntry(resourceWrapper, null, context);
 }

 // 創建當前條目
 Entry e = new CtEntry(resourceWrapper, chain, context);
 try {
 // 觸發責任鏈（從第一個開始執行到最后一個責任鏈節點，主要有創建節點、統計指標、驗證各種規則...）
 chain.entry(context, resourceWrapper, null, count, prioritized, args);
 } catch (BlockException e1) {
 // 被阻塞后退出當前條目，並統計指標
 e.exit(count, args);
 throw e1;
 } catch (Throwable e1) {
 // This should not happen, unless there are errors existing in Sentinel internal.
 RecordLog.info("Sentinel unexpected exception", e1);
 }
 return e;
}

責任鏈模式

以上 entryWithPriority 源碼中可以 sentinel 用到了責任鏈模式，通過責任鏈創建節點、統計指標、驗證規則…。
接下看下 Sentinel 是如何實現責任鏈模式又是如何統計指標和驗證規則的。

1
2
3
4
5
6
7
 
// 在沒有調用鏈，並且調用鏈沒有超過最大允許數時，初始化一個
chain = SlotChainProvider.newSlotChain();
Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(
 chainMap.size() + 1);
newMap.putAll(chainMap);
newMap.put(resourceWrapper, chain);
chainMap = newMap;
 
1
2
3
 
// 獲取到一個默認的slot調用鏈構建器，並開始構建
slotChainBuilder = SpiLoader.loadFirstInstanceOrDefault(SlotChainBuilder.class, DefaultSlotChainBuilder.class);
slotChainBuilder.build();
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
 
public ProcessorSlotChain build() {
 // 創建調用鏈對象
 ProcessorSlotChain chain = new DefaultProcessorSlotChain();

 // Note: the instances of ProcessorSlot should be different, since they are not stateless.
 // 通過SPI發現並加載並排序所有的調用鏈節點
 List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class);
 for (ProcessorSlot slot : sortedSlotList) {
 if (!(slot instanceof AbstractLinkedProcessorSlot)) {
 RecordLog.warn("The ProcessorSlot(" + slot.getClass().getCanonicalName() + ") is not an instance of AbstractLinkedProcessorSlot, can't be added into ProcessorSlotChain");
 continue;
 }
 // 按順序依次將調用鏈節點添加都最后一個，並關聯下一個節點
 chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
 }

 return chain;
}
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
public static <T> List<T> loadPrototypeInstanceListSorted(Class<T> clazz) {
 try {
 // @1
 // Not use SERVICE_LOADER_MAP, to make sure the instances loaded are different.
 ServiceLoader<T> serviceLoader = ServiceLoaderUtil.getServiceLoader(clazz);

 List<SpiOrderWrapper<T>> orderWrappers = new ArrayList<>();
 for ( T spi : serviceLoader ) {
 // @2
 int order = SpiOrderResolver.resolveOrder(spi);
 
 // @3
 // Since SPI is lazy initialized in ServiceLoader, we use online sort algorithm here.
 SpiOrderResolver.insertSorted(orderWrappers, spi, order);
 RecordLog.debug("[SpiLoader] Found {} SPI: {} with order {}", clazz.getSimpleName(),
 spi.getClass().getCanonicalName(), order);
 }
 List<T> list = new ArrayList<>(orderWrappers.size());
 // @4
 for (int i = 0; i < orderWrappers.size(); i++) {
 list.add(orderWrappers.get(i).spi);
 }
 return list;
 } catch (Throwable t) {
 RecordLog.error("[SpiLoader] ERROR: loadPrototypeInstanceListSorted failed", t);
 t.printStackTrace();
 return new ArrayList<>();
 }
}
  • @1 SPI 發現並加載ProcessorSlot接口對象集合。通過[META-INF/services/com.alibaba.csp.sentinel.slotchain.ProcessorSlot]找到所有的調用鏈節點
  • @2 每個實現類上都有一個注解 @SpiOrder 取出注解上的值，用於后續的排序
  • @3 按 @SpiOrder 從小到大冒泡排序，將 spi 插入到 orderWrappers 中
  • @4 創建一個新的集合並將 spi 按順序存入

在完成以上步驟后，調用鏈將被初始化成

順序 節點 作用 下一個節點
1 DefaultProcessorSlotChain 第一個節點 NodeSelectorSlot
2 NodeSelectorSlot 創建當前Node ClusterBuilderSlot
3 ClusterBuilderSlot 創建全局Cluster節點 LogSlot
4 LogSlot 記錄日志 StatisticSlot
5 StatisticSlot 統計各項指標 AuthoritySlot
6 AuthoritySlot 驗證認證規則 SystemSlot
7 SystemSlot 驗證系統指標（CPU等指標） FlowSlot
8 FlowSlot 驗證限流指標 DegradeSlot
9 DegradeSlot 驗證熔斷指標 Null

責任鏈調用

NodeSelectorSlot 源碼分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
 
DefaultNode node = map.get(context.getName());
if (node == null) {
 synchronized (this) {
 node = map.get(context.getName());
 if (node == null) {
 node = new DefaultNode(resourceWrapper, null);
 HashMap<String, DefaultNode> cacheMap = new HashMap<String, DefaultNode>(map.size());
 cacheMap.putAll(map);
 cacheMap.put(context.getName(), node);
 map = cacheMap;
 // Build invocation tree
 ((DefaultNode) context.getLastNode()).addChild(node);
 }

 }
}

context.setCurNode(node);
fireEntry(context, resourceWrapper, node, count, prioritized, args);

NodeSelectorSlot 源碼比較簡單，主要邏輯就是根據 context 名找到一個對應的 Node 如果沒有就創建一個，並標記為 context 的
當前 node

ClusterBuilderSlot 源碼分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 
if (clusterNode == null) {
 synchronized (lock) {
 if (clusterNode == null) {
 // Create the cluster node.
 clusterNode = new ClusterNode(resourceWrapper.getName(), resourceWrapper.getResourceType());
 HashMap<ResourceWrapper, ClusterNode> newMap = new HashMap<>(Math.max(clusterNodeMap.size(), 16));
 newMap.putAll(clusterNodeMap);
 newMap.put(node.getId(), clusterNode);

 clusterNodeMap = newMap;
 }
 }
}
node.setClusterNode(clusterNode);
  • clusterNode 是相對資源唯一
  • 因為一個資源只會有一個責任鏈，只有在初始化的時候需要進行緩存，所以這里只需要用 HashMap 用來存儲這個 clusterNode， 並且在初始化的時候加上鎖就可以了（后續只會讀）。

LogSlot 源碼分析

1
2
3
4
5
6
7
8
9
10
11
12
 
try {
 // @1
 fireEntry(context, resourceWrapper, obj, count, prioritized, args);
} catch (BlockException e) {
 // @2
 EagleEyeLogUtil.log(resourceWrapper.getName(), e.getClass().getSimpleName(), e.getRuleLimitApp(),
 context.getOrigin(), count);
 throw e;
} catch (Throwable e) {
 // @3
 RecordLog.warn("Unexpected entry exception", e);
}
  • @1 先調用后面的責任鏈節點
  • @2 當后面的責任鏈節點觸發 BlockException 異常后記錄 Block 次數到鷹眼
  • @3 當后面的責任鏈觸發其他異常后打出警告日志

StatisticSlot 源碼分析

StatisticSlot 是 Sentinel 核心的一個類，統計各項指標用於后續的限流、熔斷、系統保護等策略，接下來看下 Sentinel 是如何通過 StatisticSlot 進行指標統計的

1
2
3
4
5
6
7
8
9
10
 
// ...省略部分代碼
// Do some checking.
// @1
fireEntry(context, resourceWrapper, node, count, prioritized, args);

// Request passed, add thread count and pass count.
// @2
node.increaseThreadNum();
node.addPassRequest(count);
// ...省略部分代碼
  • @1 觸發后面的責任鏈節點
  • @2 記錄通過的線程數+1和通過請求 +count
    這里的 node 就是第二個責任鏈節點 NodeSelectorSlot 創建的 DefaultNode
    在分析源碼前可以先簡單了解下 Context、Entry、DefaultNode、ClusterNode 的關系
    Context 關系圖
  • Context 每個線程是獨享的，但是不同線程的 Context 可以使用同一個名字
  • EntranceNode 是根據 Context 名共享的，也就是說一個 Context.name 對應一個 EntranceNode。每次調用的時候都會創建，用於記錄
  • Entry 是相對於每個 Context 獨享的即是同一個 Context.name，包含了資源名、curNode（當前統計節點）、originNode（來源統計節點）等信息
  • DefaultNode 一個 Context.name 對應一個統計某資源調用鏈路上的指標
  • ClusterNode 一個資源對應一個，統計一個資源維度的指標
  • DefaultNode 和 ClusterNode 都繼承至 StatisticNode 都包含兩個 ArrayMetric 類型的字段 rollingCounterInSecond、rollingCounterInMinute 分別用於存儲秒級和分鍾級統計指標
  • 而 ArrayMetric 類包含了一個 LeapArray<MetricBucket> 類型字段 data, data 中存放了一個 WindowWrap<MetricBucket> 元素的數組（滑動窗口）, 而這個數組就是各項指標最終存儲的位置
1
 
node.increaseThreadNum();

這行代碼其實就是對 StatisticNode.curThreadNum 進行自增操作

1
2
3
4
 
public void addPassRequest(int count) {
 super.addPassRequest(count);
 this.clusterNode.addPassRequest(count);
}

添加通過的數量， 除了 DefaultNode 記錄一次外，在 ClusterNode 上也需要記錄一次【注意：ClusterNode 是按照資源維度統計的，這里指向的 ClusterNode 與同一資源不同 Context 指向的 ClusterNode 是同一個】。一個 Node 在調用了 addPassRequest
后發生了什么呢？

1
2
3
4
 
public void addPassRequest(int count) {
 rollingCounterInSecond.addPass(count);
 rollingCounterInMinute.addPass(count);
}

在以上代碼可以看到 rollingCounterInSecond 、rollingCounterInMinute 兩個字段，它們分別用來統計秒級指標和分鍾級指標。而實際上這兩個字段使用了滑動時間窗口數據結構用於存儲指標。接下來看下 Sentinel 滑動窗口的設計:
時間滑動窗口主要用到的幾個類有：

  • ArrayMetric: 負責初始化時間滑動窗口和維護
  • LeapArray: 一個滑動時間窗口主體
  • WindowWrap: 一個時間窗口主體
  • LongAdder：指標統計的計數類

ArrayMetric 構造器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
 
public ArrayMetric(int sampleCount, int intervalInMs) {
 this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
}

public ArrayMetric(int sampleCount, int intervalInMs, boolean enableOccupy) {
 if (enableOccupy) {
 this.data = new OccupiableBucketLeapArray(sampleCount, intervalInMs);
 } else {
 this.data = new BucketLeapArray(sampleCount, intervalInMs);
 }
}

/**
 * For unit test.
 */
public ArrayMetric(LeapArray<MetricBucket> array) {
 this.data = array;
}

ArrayMetric 主要有三種構造器，最后一種只是用來跑單元測試使用，而前兩種構造器主要為了初始化 data 字段。
從代碼中我們可以看到 LeapArray 有兩種實現方式 OccupiableBucketLeapArray 和 BucketLeapArray，而兩種都繼承至 LeapArray。

LeapArray 類圖
LeapArray類圖
LeapArray 類主要包含以下幾個字段：

  • int windowLengthInMs 一個時間窗口的長度，用毫秒表示
  • int sampleCount 表示用幾個時間窗口統計
  • int intervalInMs 輪回時間，也就是所有時間窗口加起來的總時長
  • AtomicReferenceArray<WindowWrap<T>> array 時間窗口實例集合，數組的長度等於 sampleCount

那么我們在回頭看下 rollingCounterInSecond 、rollingCounterInMinute 用到了哪種 LeapArray

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 
/**
 * SampleCountProperty.SAMPLE_COUNT = 2
 * IntervalProperty.INTERVAL = 1000
 * Holds statistics of the recent {@code INTERVAL} seconds. The {@code INTERVAL} is divided into time spans
 * by given {@code sampleCount}.
 */
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT,
 IntervalProperty.INTERVAL);

/**
 * Holds statistics of the recent 60 seconds. The windowLengthInMs is deliberately set to 1000 milliseconds,
 * meaning each bucket per second, in this way we can get accurate statistics of each second.
 */
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);

從上述代碼中我們可以看到秒級統計初始化了一個 OccupiableBucketLeapArray 輪回時間為 1000ms 也就是 1s，分兩個時間窗口每個各 500ms，而分鍾級統計初始化了 BucketLeapArray 輪回時間為 60000ms 也就是 1Min ，分 60 個時間窗口每個窗口 1s。

1
2
3
4
5
 
// ArrayMetric.addPass
public void addPass(int count) {
 WindowWrap<MetricBucket> wrap = data.currentWindow();
 wrap.value().addPass(count);
}

在添加通過指標前先獲取到當前的時間窗口，再將通過數量統計到窗口對應的 MetricBucket 中，那么如何獲取當前窗口呢？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
public WindowWrap<T> currentWindow() {
 return currentWindow(TimeUtil.currentTimeMillis());
}

public WindowWrap<T> currentWindow(long timeMillis) {
 if (timeMillis < 0) {
 return null;
 }

 // private int calculateTimeIdx(long timeMillis) {
 // long timeId = timeMillis / windowLengthInMs;
 // // Calculate current index so we can map the timestamp to the leap array.
 // return (int)(timeId % array.length());
 // }
 int idx = calculateTimeIdx(timeMillis);
 // Calculate current bucket start time.
 long windowStart = calculateWindowStart(timeMillis);

 /*
 * Get bucket item at given time from the array.
 *
 * (1) Bucket is absent, then just create a new bucket and CAS update to circular array.
 * (2) Bucket is up-to-date, then just return the bucket.
 * (3) Bucket is deprecated, then reset current bucket and clean all deprecated buckets.
 */
 while (true) {
 WindowWrap<T> old = array.get(idx);
 if (old == null) {
 WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
 if (array.compareAndSet(idx, null, window)) {
 return window;
 } else {
 Thread.yield();
 }
 } else if (windowStart == old.windowStart()) {
 return old;
 } else if (windowStart > old.windowStart()) {
 if (updateLock.tryLock()) {
 try {
 // Successfully get the update lock, now we reset the bucket.
 return resetWindowTo(old, windowStart);
 } finally {
 updateLock.unlock();
 }
 } else {
 // Contention failed, the thread will yield its time slice to wait for bucket available.
 Thread.yield();
 }
 } else if (windowStart < old.windowStart()) {
 // Should not go through here, as the provided time is already behind.
 return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
 }
 }
}

第一步首先獲取到當前的時間戳毫秒，通過時間戳計算出時間窗口數組的下標。在計算下標時首先將當前時間戳除以單個窗口時長，計算出當前所在從0ms開始到現在的第幾個窗，再對窗口數取模得出當前窗口的在數組中所在下標。從這里我們大概可以看出，這里數組中的時間窗口對象是反復使用的只是代表的時間不同了。
我們以秒級統計為例模擬計算下，當前時間戳為：1595495124658，按照 timeMillis / windowLengthInMs 可以得出 timeId 為 3190990249。 (int)(timeId % array.length()) 就是 3190990249 % 2 算出結果為 1，也就是說 1 下標位置的時間窗口是當前時間窗口。

第二步在計算出當前窗口所在下標后，需要計算出當前窗口的開始時間 timeMillis - timeMillis % windowLengthInMs，timeMillis % windowLengthInMs 表示當前窗口開始時間到當前時間的時長，所有當前時間減去時長即可得出當前窗口的開始時間，按上面的例子算出的結果就是 1595495124500

1
2
3
4
5
6
7
8
9
 
WindowWrap<T> old = array.get(idx);
if (old == null) {
 WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
 if (array.compareAndSet(idx, null, window)) {
 return window;
 } else {
 Thread.yield();
 }
}

第三步根據下標取出我們的當前窗口的實例，如果實例還沒有被創建過新建一個窗口實例並初始化同時通過 CAS 的方式更新到窗口數組中，如果更新失敗讓出 CPU 等待下次 CPU 執行本線程。

第四步如果下標位置已經存在一個窗口實例，並且窗口的開始時間跟本次窗口開始時間一致（同一個窗口），直接返回下標中的窗口

第五步如果當前窗口的開始時間大於下標窗口的開始時間，說明下標窗口已過期，需要重置數組下標中的窗口（把下標窗口的開始時間改完當前窗口時間，並將指標計數都置成 0 ）

第六步當前窗口時間小於下標窗口時間，重新實例化一個窗口（不太有這個可能，sentinel 內部實現了自己的時間戳）

在拿到當前時間所在窗口后，將當前的指標累加記錄到 MetriBucket 中

  • MetriBucket 累加通過指標 *
1
2
3
4
5
6
7
8
 
public void addPass(int n) {
 add(MetricEvent.PASS, n);
}

public MetricBucket add(MetricEvent event, long n) {
 counters[event.ordinal()].add(n);
 return this;
}
  • counters 是一個 LongAdder 類型的數組
  • MetricEvent 是指標類型，分別有：PASS 通過、BLOCK 阻塞、 EXCEPTION 異常、 SUCCESS 成功、 RT 平均響應時間、 OCCUPIED_PASS 通過未來的配額
  • counters[event.ordinal()].add(n) 在指定的指標計數器上累加計數

看到這里我們知道了 pass 指標是在資源通過 StatisticSlot 后幾個節點的驗證后立即進行指標計數，那么剩下的 BLOCK、 EXCEPTION、 SUCCESS、 RT、 OCCUPIED_PASS 這幾個是在什么時候做記錄的呢?

BLOCK 統計
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
 
...省略部分代碼...
} catch (BlockException e) {
 ...省略部分代碼...
 // Add block count.
 node.increaseBlockQps(count);
 if (context.getCurEntry().getOriginNode() != null) {
 context.getCurEntry().getOriginNode().increaseBlockQps(count);
 }

 if (resourceWrapper.getEntryType() == EntryType.IN) {
 // Add count for global inbound entry node for global statistics.
 Constants.ENTRY_NODE.increaseBlockQps(count);
 }

 // Handle block event with registered entry callback handlers.
 for (ProcessorSlotEntryCallback<DefaultNode> handler : StatisticSlotCallbackRegistry.getEntryCallbacks()) {
 handler.onBlocked(e, context, resourceWrapper, node, count, args);
 }

 throw e;
}

在后續的責任鏈節點中（StatisticSlot 之后的節點），如果捕獲到了阻塞異常，將對 DefaultNode、OriginNode、ENTRY_NODE 幾個 node 進行指標累計。同樣也是添加到當前窗口 MetricBucket 中不再進行過多描述

EXCEPTION 統計
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
 
try {
 // Do some checking.
 fireEntry(context, resourceWrapper, node, count, prioritized, args);
 ...省略部分代碼
} catch (Throwable e) {
 // Unexpected error, set error to current entry.
 context.getCurEntry().setError(e);

 // This should not happen.
 node.increaseExceptionQps(count);
 if (context.getCurEntry().getOriginNode() != null) {
 context.getCurEntry().getOriginNode().increaseExceptionQps(count);
 }

 if (resourceWrapper.getEntryType() == EntryType.IN) {
 Constants.ENTRY_NODE.increaseExceptionQps(count);
 }
 throw e;
}

類似的 exception 統計在后續的責任鏈節點中（StatisticSlot 之后的節點），如果捕獲到了異常，將對 DefaultNode、OriginNode、ENTRY_NODE 幾個 node 進行指標累計。

除了 StatisticSlot 自動捕獲異常外，在資源調用過程中如果出現了異常將通過調用 Tracer.trace(e) 手動統計異常指標

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
 
public static void trace(Throwable e, int count) {
 traceContext(e, count, ContextUtil.getContext());
}
public static void traceContext(Throwable e, int count, Context context) {
 if (!shouldTrace(e)) {
 return;
 }

 if (context == null || context instanceof NullContext) {
 return;
 }

 DefaultNode curNode = (DefaultNode)context.getCurNode();
 traceExceptionToNode(e, count, context.getCurEntry(), curNode);
}

首先從線程變量中出去當前線程的 Context 在從中取出 DefaultNode 和 ClusterNode 並進行異常指標累計

SUCCESS、 RT 統計

平均響應時間和成功次數的統計是在資源退出的時候調用 entry.exit() 進行統計，代碼如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
// StatisticSlot#exit()
public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
 DefaultNode node = (DefaultNode)context.getCurNode();

 if (context.getCurEntry().getError() == null) {
 // Calculate response time (max RT is statisticMaxRt from SentinelConfig).
 long rt = TimeUtil.currentTimeMillis() - context.getCurEntry().getCreateTime();
 int maxStatisticRt = SentinelConfig.statisticMaxRt();
 if (rt > maxStatisticRt) {
 rt = maxStatisticRt;
 }

 // Record response time and success count.
 node.addRtAndSuccess(rt, count);
 if (context.getCurEntry().getOriginNode() != null) {
 context.getCurEntry().getOriginNode().addRtAndSuccess(rt, count);
 }

 node.decreaseThreadNum();

 if (context.getCurEntry().getOriginNode() != null) {
 context.getCurEntry().getOriginNode().decreaseThreadNum();
 }

 if (resourceWrapper.getEntryType() == EntryType.IN) {
 Constants.ENTRY_NODE.addRtAndSuccess(rt, count);
 Constants.ENTRY_NODE.decreaseThreadNum();
 }
 } else {
 // Error may happen.
 }

 // Handle exit event with registered exit callback handlers.
 Collection<ProcessorSlotExitCallback> exitCallbacks = StatisticSlotCallbackRegistry.getExitCallbacks();
 for (ProcessorSlotExitCallback handler : exitCallbacks) {
 handler.onExit(context, resourceWrapper, count, args);
 }

 fireExit(context, resourceWrapper, count);
}

退出也是責任鏈調用退出每個節點，這里直接跳過了大部分代碼。退出統計大致流程如下：

  • 獲取得到當前時間戳和資源調用的時間，相減算出這次整個資源調用所花費的總時間
  • 將總時間記錄和成功次數累加記錄當前窗口，本次總時間如果超過最大統計時間以最大統計時間作為本次統計時間
  • 對 Node 扣減一次當前線程數
  • 觸發下一個責任鏈節點退出

LongAdder 源碼分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
 
public void add(long x) {
 Cell[] as = cells;
 long b = base;
 long v;
 HashCode hc;
 Cell a;
 int n;
 if (cells != null || !casBase(base, b + x)) {
 boolean uncontended = true;
 hc = threadHashCode.get()
 int h = hc.code;
 n = as.length;
 a = as[(n - 1) & h]
 uncontended = a.cas(v = a.value, v + x)
 if (as == null || as.length < 1 ||
 a == null ||
 !uncontended) {
 retryUpdate(x, hc, uncontended);
 }
 }
}

LongAdder 中有一個Cell數組用於存儲數值，當高並發時對數組中某個值進行加法運算減少同一個數值並發。（+1） 或者 （+ -1）

1
2
3
4
5
6
7
8
9
10
11
12
 
public long sum() {
 long sum = base;
 Cell[] as = cells;
 if (as != null) {
 int n = as.length;
 for (int i = 0; i < n; ++i) {
 Cell a = as[i];
 if (a != null) { sum += a.value; }
 }
 }
 return sum;
}

取值時把 Cell 數組中所有元素的取出算總數

熔點判斷

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 
DegradeRuleManager.checkDegrade(resourceWrapper, context, node, count);

public static void checkDegrade(ResourceWrapper resource, Context context, DefaultNode node, int count)
 throws BlockException {

 Set<DegradeRule> rules = degradeRules.get(resource.getName());
 if (rules == null) {
 return;
 }

 for (DegradeRule rule : rules) {
 if (!rule.passCheck(context, node, count)) {
 throw new DegradeException(rule.getLimitApp(), rule);
 }
 }
}

熔點的判斷是由 DegradeRuleManager 管理。 DegradeRuleManager 會根據資源名取出所有的熔斷規則，然后檢查所有的規則如果觸發其中一個直接拋出 DegradeException 異常觸發熔斷機制。

  • RT *
1
2
3
4
5
6
7
8
9
10
11
12
13
 
double rt = clusterNode.avgRt();
if (rt < this.count) {
 passCount.set(0); // 計數，用於判斷連續超 RT 多少次
 return true;
}

// Sentinel will degrade the service only if count exceeds.
if (passCount.incrementAndGet() < rtSlowRequestAmount) {
 return true;
}

...省略部分代碼
return false;
  • 從 clusterNode 中計算出平均響應時間
  • 如果平均響應時間小於規則設置時間，將統計連續超時計數器重置為0
  • 如果平均響應時間大於規則設置時間，並且連續超時計數器超過了規則設置的大小，判為到達熔斷點拋出熔斷異常

統計平均 RT 的方法（秒級）：

  • 取出所有窗口（秒級只定義了兩個時間窗口）的 RT，並求總和
  • 取出所有窗口（秒級只定義了兩個時間窗口）的 success，並求總和
  • 所有窗口的 RT 總和 除以 success 總和 得出平均RT

異常比例熔斷也是類似的邏輯（秒級）

  • 取出所有窗口的 exception 數求和，並除以一個間隔時間（秒為單位）【每秒總異常數】
  • 取出所有窗口的 success 數求總和，並除以一個間隔時間（秒為單位）【每秒總退出成功數，包含了異常數】
  • 取出所有窗口的 pass 總和 加上所有窗口 block 總數，並除以一個間隔時間（秒為單位）【算每秒總調用量】
  • 如果每秒總調用量小於 minRequestAmount 判為未到達熔斷點
  • 如果每秒總異常數沒有超過 minRequestAmount 判為未到達熔斷點
  • 每秒總退出成功數 / 每秒總異常數（異常比例）如果超過規則指定比例，判為到達熔斷點拋出熔斷異常

異常數就比例（分鍾級）

  • 取出所有窗口的 exception 數總和，判斷如果超過規則配置數，拋出熔斷異常

總結

Sentinel 通過責任鏈，觸發節點創建、監控統計、日志、認證、系統限流、限流、熔斷，因為Sentinl 是由 SPI 創建的責任鏈所以我們可以自定義鏈節點拿到指標根據自己的業務邏輯定義。
Sentinel 通過將所有的指標統計到時間窗口中，記錄在 MetricBucket 類實例中


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 Sentinel之熔斷降級 Sentinel限流與熔斷分析 Sentinel熔斷降級 Sentinel實現熔斷和限流 Sentinel 熔斷降級 Sentinel流控、熔斷解析 Sentinel隔離和降級-熔斷策略 阿里熔斷限流Sentinel研究 Springcloudalibaba之Sentinel服務熔斷與限流 使用Sentinel實現熔斷降級
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM