Sentinel 源码学习

2024-02-20 12:45 由废物大师兄发表于 #后端开发

引入依赖

<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：相当于将窗口向后移动）