Sentinel 源碼學習_ZenDei技術網路在線

引入依賴 <dependency> <groupId>com.alibaba.csp</groupId> <artifactId>sentinel-core</artifactId> <version>1.8.7</version> </dependency> 基本用法 try (Entry ent ...

引入依賴

<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-core</artifactId>
    <version>1.8.7</version>
</dependency>

基本用法

try (Entry entry = SphU.entry("HelloWorld")) {
    // 被保護的邏輯
    System.out.println("hello world");
} catch (BlockException ex) {
    // 處理被流控的邏輯
    System.out.println("blocked!");
}

接下來，閱讀源碼，我們從SphU.entry()開始

每個SphU#entry()將返回一個Entry。這個類維護了當前調用的一些信息：

createTime ：這個entry的創建時間，用於響應時間統計
current Node ：在當前上下文中的資源的統計
origin Node ：原始節點的統計
ResourceWrapper ：資源名稱

CtSph#entryWithPriority()方法就是整個流控的基本流程：

1、首先，獲取當前線程上下文，如果為空，則創建一個

2、然後，查找處理器鏈

3、最後，依次執行處理器

這是一個典型的責任鏈

接下來，挨個來看，首先看一下上下文。上下文是一個線程局部變數 ThreadLocal<Context>

如果當前線程還沒有上下文，則創建一個

有了Context之後，接下來查找處理器

這些功能插槽（slot chain）有不同的職責：

NodeSelectorSlot ：負責收集資源的路徑，並將這些資源的調用路徑，以樹狀結構存儲起來，用於根據調用路徑來限流降級；
ClusterBuilderSlot ：用於存儲資源的統計信息以及調用者信息，例如該資源的 RT, QPS, thread count 等等，這些信息將用作為多維度限流，降級的依據；
StatisticSlot ：用於記錄、統計不同緯度的 runtime 指標監控信息；
FlowSlot ：用於根據預設的限流規則以及前面 slot 統計的狀態，來進行流量控制；
AuthoritySlot ：根據配置的黑白名單和調用來源信息，來做黑白名單控制；
DegradeSlot ：通過統計信息以及預設的規則，來做熔斷降級；
SystemSlot ：通過系統的狀態，例如 load1 等，來控制總的入口流量；

到這裡為止，資源有了，上下文有了，處理器鏈有了，於是，接下來就可以對資源應用所有的處理器了

關於功能插槽的學習就先到這裡，下麵補充一個知識點：Node

Node 用於保存資源的實時統計信息

StatisticNode 保存三種實時統計指標：

秒級指標
分鐘級指標
線程數

DefaultNode 用於保存特定上下文中特定資源名稱的統計信息

EntranceNode 代表調用樹的入口

總之一句話，Node是用於保存統計信息的。那麼，這些指標數據是如何計數的呢？

Sentinel 使用滑動視窗實時記錄和統計資源指標。ArrayMetric背後的滑動視窗基礎結構是LeapArray。

下麵重點看一下StatisticNode

StatisticNode是用於實時統計的處理器插槽。在進入這個槽位時，需要分別計算以下信息：

ClusterNode ：該資源ID的集群節點統計信息總和
Origin node ：來自不同調用者/起源的集群節點的統計信息
DefaultNode ：特定上下文中特定資源名稱的統計信息
最後，是所有入口的總和統計

private int calculateTimeIdx(/*@Valid*/ long timeMillis) {
    long timeId = timeMillis / windowLengthInMs;
    // Calculate current index so we can map the timestamp to the leap array.
    return (int)(timeId % array.length());
}

protected long calculateWindowStart(/*@Valid*/ long timeMillis) {
    return timeMillis - timeMillis % windowLengthInMs;
}

/**
 * Get bucket item at provided timestamp.
 *
 * @param timeMillis a valid timestamp in milliseconds
 * @return current bucket item at provided timestamp if the time is valid; null if time is invalid
 */
public WindowWrap<T> currentWindow(long timeMillis) {
    if (timeMillis < 0) {
        return null;
    }

    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.
     */
    while (true) {
        WindowWrap<T> old = array.get(idx);
        if (old == null) {
            /*
             *     B0       B1      B2    NULL      B4
             * ||_______|_______|_______|_______|_______||___
             * 200     400     600     800     1000    1200  timestamp
             *                             ^
             *                          time=888
             *            bucket is empty, so create new and update
             *
             * If the old bucket is absent, then we create a new bucket at {@code windowStart},
             * then try to update circular array via a CAS operation. Only one thread can
             * succeed to update, while other threads yield its time slice.
             */
            WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
            if (array.compareAndSet(idx, null, window)) {
                // Successfully updated, return the created bucket.
                return window;
            } else {
                // Contention failed, the thread will yield its time slice to wait for bucket available.
                Thread.yield();
            }
        } else if (windowStart == old.windowStart()) {
            /*
             *     B0       B1      B2     B3      B4
             * ||_______|_______|_______|_______|_______||___
             * 200     400     600     800     1000    1200  timestamp
             *                             ^
             *                          time=888
             *            startTime of Bucket 3: 800, so it's up-to-date
             *
             * If current {@code windowStart} is equal to the start timestamp of old bucket,
             * that means the time is within the bucket, so directly return the bucket.
             */
            return old;
        } else if (windowStart > old.windowStart()) {
            /*
             *   (old)
             *             B0       B1      B2    NULL      B4
             * |_______||_______|_______|_______|_______|_______||___
             * ...    1200     1400    1600    1800    2000    2200  timestamp
             *                              ^
             *                           time=1676
             *          startTime of Bucket 2: 400, deprecated, should be reset
             *
             * If the start timestamp of old bucket is behind provided time, that means
             * the bucket is deprecated. We have to reset the bucket to current {@code windowStart}.
             * Note that the reset and clean-up operations are hard to be atomic,
             * so we need a update lock to guarantee the correctness of bucket update.
             *
             * The update lock is conditional (tiny scope) and will take effect only when
             * bucket is deprecated, so in most cases it won't lead to performance loss.
             */
            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));
        }
    }
}

現在，有2個視窗，每個視窗500ms，2個視窗總共1000ms

假設，當前時間戳是1200ms，那麼 (1200 / 500) % 2 = 0， 1200 - 1200 % 500 = 1000

這個時候，如果0這個位置沒有視窗，則創建一個新的視窗，新視窗的視窗開始時間是1000ms

如果0這個位置有視窗，則繼續判斷舊視窗的視窗開始時間是否為1000ms，如果是，則表示視窗沒有過期，直接返回該視窗。如果舊視窗的開始時間小於1000ms，則表示舊視窗過期了，於是重置舊視窗的統計數據，重新設置視窗開始時間（PS：相當於將視窗向後移動）