Sentinel原理解析

启动

触发方式

SphU.entry("自定义资源名")

public static Entry entry(String name) throws BlockException {
    return Env.sph.entry(name, EntryType.OUT, 1, OBJECTS0);
}
  1. 执行 Env 静态代码块
  2. 进入 CtSph#entry 方法

Env静态代码块

在 InitExecutor#doInit 中，通过SPI机制加载所有 InitFunc 实现类，然后按顺序调用他们的 init() 方法

public class Env {
    public static final Sph sph = new CtSph();
    static {
        // If init fails, the process will exit.
        InitExecutor.doInit();
    }
}

常见的 InitFunc 实现类

  1. com.alibaba.csp.sentinel.metric.extension.MetricCallbackInit 统计Metric信息
  2. com.alibaba.csp.sentinel.transport.init.CommandCenterInitFunc transport相关
  3. com.alibaba.csp.sentinel.transport.init.HeartbeatSenderInitFunc transport相关

CtSph#entry

  1. 基于 name、type 包装一个 StringResourceWrapper 对象，即抽象的资源；

  2. 进入 CtSph#entryWithPriority 方法，

  3. 创建一个默认的 Context 对象
    InternalContextUtil.internalEnter(Constants.CONTEXT_DEFAULT_NAME) =>
    ContextUtil#trueEnter =>

protected static Context trueEnter(String name, String origin) {
    // 从 ThreadLocal 中获取
    Context context = contextHolder.get();
    if (context == null) {
        // contextNameNodeMap： key -> DefaultNode , key为contextName , value为EntranceNode
        Map<String, DefaultNode> localCacheNameMap = contextNameNodeMap;
        DefaultNode node = localCacheNameMap.get(name);
        if (node == null) {
            // Context 最大值为 2000
            if (localCacheNameMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {
                setNullContext();
                return NULL_CONTEXT;
            } else {
                try {
                    LOCK.lock();
                    node = contextNameNodeMap.get(name);
                    if (node == null) {
                        if (contextNameNodeMap.size() > Constants.MAX_CONTEXT_NAME_SIZE) {
                            setNullContext();
                            return NULL_CONTEXT;
                        } else {
                            // 创建一个 EntranceNode
                            node = new EntranceNode(new StringResourceWrapper(name, EntryType.IN), null);
                            // 添加到 根节点
                            Constants.ROOT.addChild(node);

                            Map<String, DefaultNode> newMap = new HashMap<>(contextNameNodeMap.size() + 1);
                            newMap.putAll(contextNameNodeMap);
                            newMap.put(name, node);
                            contextNameNodeMap = newMap;
                        }
                    }
                } finally {
                    LOCK.unlock();
                }
            }
        }
        // 基于 EntranceNode 创建 Context , 保存到 ThreadLocal 并返回
        context = new Context(node, name);
        context.setOrigin(origin);
        contextHolder.set(context);
    }
    return context;
}

Slot链

ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
    ProcessorSlotChain chain = chainMap.get(resourceWrapper);
    // 双重校验
    if (chain == null) {
        synchronized (LOCK) {
            chain = chainMap.get(resourceWrapper);
            if (chain == null) {
                // 每个资源对应一个 Slot链 ， 资源数最大为 6000
                if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {
                    return null;
                }

                // 先通过SPI获取ProcessorSlotChain， 如果没有返回默认的 DefaultSlotChainBuilder
                chain = SlotChainProvider.newSlotChain();
                Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(
                    chainMap.size() + 1);
                newMap.putAll(chainMap);
                newMap.put(resourceWrapper, chain);
                chainMap = newMap;
            }
        }
    }
    return chain;
}

DefaultSlotChainBuilder 中添加了一系列的 Solt , 各个 Solt 执行的顺序，就是创建时添加的顺序：

public class DefaultSlotChainBuilder implements SlotChainBuilder {
    @Override
    public ProcessorSlotChain build() {
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();
        chain.addLast(new NodeSelectorSlot());
        chain.addLast(new ClusterBuilderSlot());
        chain.addLast(new LogSlot());
        chain.addLast(new StatisticSlot());
        chain.addLast(new AuthoritySlot());
        chain.addLast(new SystemSlot());
        chain.addLast(new FlowSlot());
        chain.addLast(new DegradeSlot());
        return chain;
    }
}

addLast方法主要两行代码：
1. ProcessorSlotChain.end.next = 入参Solt
2. ProcessorSlotChain.end = 入参

即最后的调用顺序如下： NodeSelectorSlot => ClusterBuilderSlot => LogSlot => StatisticSlot => AuthoritySlot => SystemSlot => FlowSlot => DegradeSlot
如果想改变他们的调用顺序，可通过SPI机制实现

NodeSelectorSlot

构造调用链，具体参考 NodeSelectorSlot

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
    throws Throwable {
    // 每个资源对应一个 ProcessorSlotChain
    // 一个资源可以对应多个Context
    // 一个ContextName 对应一个 DefaultNode , 即 一个资源可能对应多个 DefaultNode, 但 一个资源只有一个 ClusterNode
    // 针对同一段代码，不同线程对应的Context实例是不一样的，但是对应的Context Name是一样的，所以这时认为是同一个Context，Context我们用Name区分
    DefaultNode node = map.get(context.getName());
    if (node == null) {
        synchronized (this) {
            node = map.get(context.getName());
            if (node == null) {
                node = new DefaultNode(resourceWrapper, null);
                // key 为 ontextName , vaue 为 DefaultNode
                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);
}

ClusterBuilderSlot

具体参考 ClusterBuilderSlot

每个资源对应一个ClusterNode，并且DefaultNode引用了ClusterNode

LogSlot

记录日志用的，先执行下面的 Solt， 如果报错了或者被Block了，记录到日志中

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode obj, int count, boolean prioritized, Object... args)
    throws Throwable {
    try {
        fireEntry(context, resourceWrapper, obj, count, prioritized, args);
    } catch (BlockException e) {
        EagleEyeLogUtil.log(resourceWrapper.getName(), e.getClass().getSimpleName(), e.getRuleLimitApp(),
            context.getOrigin(), count);
        throw e;
    } catch (Throwable e) {
        RecordLog.warn("Unexpected entry exception", e);
    }
}

StatisticSlot

核心实现，各种计数的实现逻辑，基于时间窗口实现。 基于触发请求通过 和 请求Block 的回调逻辑，回调逻辑在 MetricCallbackInit 中初始化了， 最终还是靠 StatisticSlotCallbackRegistry

// 省略了一些代码
@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {
    try {
        // 执行下来的Solt ，判断是否通过
        fireEntry(context, resourceWrapper, node, count, prioritized, args);

        // Request passed, add thread count and pass count.
        node.increaseThreadNum();
        node.addPassRequest(count);
    } catch (BlockException e) {
        // Blocked, set block exception to current entry.
        context.getCurEntry().setError(e);

        // Add block count.
        node.increaseBlockQps(count);
        if (context.getCurEntry().getOriginNode() != null) {
            context.getCurEntry().getOriginNode().increaseBlockQps(count);
        }
    }
}

DefaultNode 继承自 StatisticNode , 在 StatisticNode 中有两个属性

// 第一个参数表示 窗口的个数；第二个参数表示 窗口对多长时间进行统计  比如 QPS xx/秒   那就是 1000 毫秒， 所以窗口的长度为  1000/个数
private transient volatile Metric rollingCounterInSecond = new ArrayMetric(SampleCountProperty.SAMPLE_COUNT, IntervalProperty.INTERVAL);

// 窗口长度为1000 60个 刚好一分钟
private transient Metric rollingCounterInMinute = new ArrayMetric(60, 60 * 1000, false);

ArrayMetric 持有 LeapArray ， LeapArray 主要有两个实现类 OccupiableBucketLeapArray 、 BucketLeapArray ， 但根据当前时间获取窗口的核心实现在 LeapArray 抽象类中

滑动窗口简单理解就是： 根据任何时间，都可以获取一个对应的窗口，在该窗口内，保存着在窗口长度时间内通过的请求数、被block的请求数、异常数、RT。基于这些数据，我们就可以得到对应的资源的QPS、RT等指标信息。

核心方法在 LeapArray#currentWindow ， 整体思路如下

  1. 根据当前时间获取时间窗口的下标 (time/windowLength) % array.length()
  2. 计算当前时间对应时间窗口的开始时间 time - time % windowLength
  3. 根据下标获取时间窗口，这里分三种情况：
    (1) 根据下标没有获取到窗口，此时创建一个窗口。此时代表窗口没有创建 或者 窗口还没有开始滑动， 所以对应的下标位置为null
    (2) 根据下标获取到窗口，并且该窗口的开始时间和上面计算的开始时间一样，此时直接返回该窗口
    (3) 根据下标获取到窗口，但是该窗口的开始时间大于上面计算的开始时间，这时需要用计算的开始时间重置该窗口的开始时间，这就类似于窗口在滑动
public WindowWrap<T> currentWindow(long timeMillis) {
    if (timeMillis < 0) {
        return null;
    }

    // 计算窗口数组下标
    int idx = calculateTimeIdx(timeMillis);
    // 计算开始时间
    long windowStart = calculateWindowStart(timeMillis);
    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)) {
                // 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()) {
            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));
        }
    }
}

todo OccupiableBucketLeapArray 还不太理解

AuthoritySlot

黑白名单规则校验，非常简单

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {
    checkBlackWhiteAuthority(resourceWrapper, context);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
  1. 加载所有的黑白名单规则
  2. 遍历所有黑白名单规则，调用 AuthorityRuleChecker#passCheck 方法，如果不通过则抛出 AuthorityException
  3. 校验逻辑：从 Context 中拿到 originName, 然后判断 originName 是否在 规则的 limitApp 中, 然后判断是 黑名单 还是白名单，然后校验返回结果

SystemSlot

仅对入口流量有效，校验顺序 QPS -> 线程数 -> RT -> BBR -> CPU

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args) throws Throwable {
    SystemRuleManager.checkSystem(resourceWrapper);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

FlowSlot

限流处理

三种拒绝策略：直接拒绝、WarnUP、匀速排队

三种限流模式：直接、关联、链路

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                    boolean prioritized, Object... args) throws Throwable {
    checkFlow(resourceWrapper, context, node, count, prioritized);

    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}

FlowRuleChecker#checkFlow

  1. 获取所有限流规则
  2. 遍历规则，执行 FlowRuleChecker#canPassCheck => FlowRuleChecker#passLocalCheck => rule.getRater().canPass(selectedNode, acquireCount, prioritized)
  3. rule.getRater() 返回一个 TrafficShapingController 对象， 它有3种实现(代码中有4中，但官方文档只介绍了3种)，即对应上面的三种流控模式，每个规则对用的 TrafficShapingController 是在加载规则的时候就确定了

// FlowRuleUtil#generateRater

private static TrafficShapingController generateRater(/*@Valid*/ FlowRule rule) {
    if (rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {
        switch (rule.getControlBehavior()) {
            case RuleConstant.CONTROL_BEHAVIOR_WARM_UP:
                return new WarmUpController(rule.getCount(), rule.getWarmUpPeriodSec(),
                    ColdFactorProperty.coldFactor);
            case RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER:
                return new RateLimiterController(rule.getMaxQueueingTimeMs(), rule.getCount());
            case RuleConstant.CONTROL_BEHAVIOR_WARM_UP_RATE_LIMITER:
                return new WarmUpRateLimiterController(rule.getCount(), rule.getWarmUpPeriodSec(),
                    rule.getMaxQueueingTimeMs(), ColdFactorProperty.coldFactor);
            case RuleConstant.CONTROL_BEHAVIOR_DEFAULT:
            default:
                // Default mode or unknown mode: default traffic shaping controller (fast-reject).
        }
    }
    return new DefaultController(rule.getCount(), rule.getGrade());
}
DefaultController

比较简单，判断逻辑 (当前的Count + 本次调用) 是否大于 规则中设置的 阈值

WarmUpController

让QPS在指定的时间内增加到 阈值， 目前每太看懂

RateLimiterController

也比较简单，先按规则中配置的QPS计算每个请求的平均响应时间，然后判断当前请求是否能够等那么久(规则中的时间窗口)

三种限流模式在哪里体现？

其实这个主要就是判断 你的指标数据应该要从哪个 Node 中获取，这部分逻辑在 FlowRuleChecker#selectNodeByRequesterAndStrategy 方法中

  1. 直接： 根据你的 originName 和 limitApp 来判断是取 ClusterNode 还是 OriginNode
  2. 关联： 根据关联的资源名取对应的 ClusterNode
  3. 链路： 判断关联的资源 和 当前的 contextName 是否一致，是则返回 当前的 DefaultNode

DegradeSlot

降级处理

目前有三种降级模式：基于RT、基于异常比例、基于一分钟异常数

@Override
public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args)
    throws Throwable {
    DegradeRuleManager.checkDegrade(resourceWrapper, context, node, count);
    fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
基于RT

从时间窗口获取RT和规则中配置的阈值进行比较， 通过则 重置计数，然后直接返回； 不通过则 计数加1，如果 计数 >= 5，则进行降级处理

基于异常比例

前提条件 QPS > =5 , 然后用 1s异常数/1s总请求数 , 和规则中配置的阈值进行比较

基于1分钟异常数

直接用1分钟内的异常数和规则的阈值做比较

如何按时间窗口降级

定时任务 + flag
如果降级了， 设置 flag = true , 在 时间窗口秒后， 重置 flag = false ，然后再 passCheck 方法的入口处， 如果 flag = true 就直接降级

和控制台交互

控制台：client端
sentinel-transport模块：Server端，有两种实现： Netty 和 Java原生ServerSocket

在我们引入sentinel-transport模块之后，就可以通过 HTTP API 来获取一些信息，例如：

http://localhost:8719/getRules?type=<XXXX>
http://localhost:8719/getParamRules

核心原理就是sentinel-transport模块启动一个http server，大概流程：

  1. client向server发送请求
  2. server端解析请求，根据url内容找到对应的CommandHandler
  3. server端执行对应的CommandHandler逻辑
  4. 将结果返回给client端

控制台 和 server 之间会维持心跳，大致流程：

  1. server向控制台发送心跳， 发送到 控制台地址/registry/machine 这个地址
  2. 控制台接收心跳，从中获取机器信息
  3. 控制台将机器信息展示到界面

Server端

源码在 sentinel-transport 模块中，分为Netty实现和Http实现

sentinel-transport-common 公用模块，被其它两个模块引用
sentinel-transport-netty-http 基于Netty实现
sentinel-transport-simple-http 基于Java原生ServerSocket实现

Common模块

在 sentinel-transport-common 的 resources/META-INFO/services 目录下，提供了两个 SPI 接口： com.alibaba.csp.sentinel.init.InitFunc 和 com.alibaba.csp.sentinel.command.CommandHandler

InitFunc之前已经介绍过了，在 Env 的静态代码块中会通过SPI机制加载所有的 InitFunc 实现类，这里主要包括两个： com.alibaba.csp.sentinel.transport.init.CommandCenterInitFunc 和 com.alibaba.csp.sentinel.transport.init.HeartbeatSenderInitFunc

CommandCenterInitFunc
该类主要负责启动Server端

  1. 通过 CommandCenterProvider 获取到优先级最高的 CommandCenter ，这部分逻辑在 CommandCenterProvider 类的静态代码块中是实现。 如果同时引入了 sentinel-transport-netty-http 和 sentinel-transport-simple-http 模块，默认 SimpleHttpCommandCenter 优先级更高
  2. 执行 CommandCenter#beforeStart 方法，该步骤主要是通过SPI加载所有 CommandHandler 实现类然后缓存起来；
  3. 执行 CommandCenter#start 方法，该步骤用于启动Server

HeartbeatSenderInitFunc
从名字上可以猜测到是和心跳检测相关的

  1. 通过 HeartbeatSenderProvider 获取优先级最高的 HeartbeatSender，这部分逻辑在 HeartbeatSenderProvider 类的静态代码块中是实现。两个模块的实现类分别对应 HttpHeartbeatSender 和 SimpleHttpHeartbeatSender
  2. 初始化 ScheduledExecutorService ，用于 定时发送心跳
  3. 设置发送心跳的间隔，全局属性 csp.sentinel.heartbeat.interval.ms
  4. 通过线程池定时执行 HeartbeatSender#sendHeartbeat 方法

CommandHandler
这个有点类似于web应用中的Controller层，不同的 CommandHandler 实现类对应不同请求url的逻辑。
而所有的 CommandHandler 实现类是在 CommandCenter#beforeStart 方法中通过SPI加载的：

Map<String, CommandHandler> handlers = CommandHandlerProvider.getInstance().namedHandlers();
基于Netty实现

主要类： NettyHttpCommandCenter 、 HttpHeartbeatSender 、 HttpServerHandler

基于ServerSocket实现

主要类： SimpleHttpCommandCenter 、 SimpleHttpHeartbeatSender 、 HttpEventTask

Client端

Server端向控制台发送心跳，控制台解析心跳包获取机器信息 ， 对应URL /registry/machine ， 即 MachineRegistryController#receiveHeartBeat 方法，获取机器信息之后添加到缓存中

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 212,294评论 6赞 493
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,493评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 157,790评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,595评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,718评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,906评论 1赞 290
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,053评论 3赞 410
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,797评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,250评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,570评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,711评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,388评论 4赞 332
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,018评论 3赞 316
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,796评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,023评论 1赞 266
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,461评论 2赞 360
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,595评论 2赞 350