聊聊服务熔断降级Sentinel

简介

分布式系统的流量防卫兵，Sentinel是面向分布式服务框架的流量控制组件，主要以流量为切入点，从限流、流量整形、熔断降级、系统负载保护、热点防护等多个维度来帮助开发者保障微服务的稳定性。Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、集群流量控制、实时熔断下游不可用应用等。Sentinel 可以简单的分为 Sentinel 核心库和 Dashboard。核心库不依赖 Dashboard，但是结合 Dashboard 可以取得最好的效果。

官网参考地址：https://github.com/alibaba/Sentinel/wiki/%E5%A6%82%E4%BD%95%E4%BD%BF%E7%94%A8

工程引入依赖

需要开启服务熔断降级的工程引入jar依赖。

<!--sentinel启动器-->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>
 
<!-- 实例化sentinel规则到nacos-->
<dependency>
    <groupId>com.alibaba.csp</groupId>
    <artifactId>sentinel-datasource-nacos</artifactId>
</dependency>

Dashboard控制台

控制台用于展示配置的规则，以及请求的实时监控，下载与sentinel对应版本的sentinel-dashboard.jar启动文件。

相关配置

下载地址：https://github.com/alibaba/Sentinel/releases；

启动命令例（设置端口、用户名、密码、jar的位置）：java -Dserver.port=8090 -Dsentinel.dashboard.auth.username=qingyun -Dsentinel.dashboard.auth.password=123456 -jar sentinel-dashboard-1.8.1.jar
工程配置文件中配置sentinel控制台连接信息：

spring:
 cloud:
   sentinel:
     transport:  
       dashboard: 127.0.0.1:8090
       port: 8719 #默认8719，假如被占用了会自动从8719开始依次+1扫描。直至找到未被占用的端口

image

Nacos实例化规则

 在Dashboard控制台上添加的规则只能存在内存中，重启控制台后规则会丢失，所以实例化规则可以选择nacos。目前项目中在nacos配置规则，在Dashboard中回显规则，监控访问情况。

nacos中配置规则

image

项目引入规则

image

Dashboard回显规则

image

Sentinel实现

注解方式定义资源

  Sentinel 支持通过 `@SentinelResource` 注解定义资源并配置 `blockHandler` 和 `fallback` 函数来进行限流之后的处理。例如：

@GetMapping("/getTestListSentinelFlow")
@SentinelResource(value = "todolist/getTestListSentinelFlow", fallback = "fallbackGetTestListSentinel",fallbackClass = FallbackAppcenter.class, blockHandler = "blockHandlerGetTestListSentinel",blockHandlerClass = BlockHandlerAppcenter.class)
public Result<List<UmappTest>> getTestListSentinelFlow() {
    Result<List<UmappTest>> result = umappTestService.getTestList();
    return result;
}
 
/**
* 处理blockHandler对应的方法
*/
public static Result<List<UmappTest>> blockHandlerGetTestListSentinel(BlockException ex) {
    return Result.error("被流控了"+ex.getMessage());
}
 
/**
* 处理fallback对应的方法
*/
public static Result<List<UmappTest>> fallbackGetTestListSentinel(Throwable e) {
    e.printStackTrace();
    return Result.error("抛出异常了"+e.getMessage());
}

参数说明

value:对应资源名
blockHandler：设置流控降级后的方法（默认改方法必须写在同一个类中），如果处理方法不想在一个类中，可以配置blockHandlerClass=xxx.class，但是方法必须是static。
fallback：当接口出现异常，可以交个fallback指定的方法处理（默认改方法必须写在同一个类中），如果处理方法不想在同一个类中，可以配置fallbackClass=xxx.class，但是方法必须是static，
blockHandler和fallback同时指定了，当触发了限流和异常时，blockHandler优先级更高
exceptionsToIgnore:数组的方式指定排除哪些异常不处理

统一异常处理

定义一个统一处理BlockException异常的类，接口名中不需要写@SentinelResource指定异常调用的方法，但是需要具体异常处理的方法，还是需要使用@SentinelResource指定。

/**
 * 统一异常处理类
 */
@Component
public class MyBlockExceptionHandler implements BlockExceptionHandler {
    Logger log = LoggerFactory.getLogger(this.getClass());
 
    @Override
    public void handle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, BlockException e) throws Exception {
        Message message = null;
        log.info("BlockException======="+e.getRule());
        if(e instanceof FlowException){
            message = message.error(300,"接口被限流了");
        } else if(e instanceof DegradeException){
            message = message.error(301,"接口被降级了");
        } else if(e instanceof ParamFlowException){
            message =  message.error(302,"接口热点参数限流了");
        } else if(e instanceof SystemBlockException){
            message =  message.error(303,"接口触发系统保护规则了");
        } else if(e instanceof AuthorityException){
            message =  message.error(304,"授权接口不通过");
        }
        //返回json格式
        httpServletResponse.setStatus(500);
        httpServletResponse.setCharacterEncoding("utf-8");
        httpServletResponse.setContentType(MediaType.APPLICATION_JSON_VALUE);
        new ObjectMapper().writeValue(httpServletResponse.getWriter(),message);
    }
}

Sentinel 功能

流量控制

每秒请求的并发数大于服务器能提供的最大QPS，对大于的部分进行限流，返回“稍后重试”或者直接拒绝处理。

image

rule-type为流控方式，例选择了flow（流控），则data-id中配置的规则都是流控，其他的规则配置无效。

规则配置参考：https://github.com/alibaba/Sentinel/wiki/%E6%B5%81%E9%87%8F%E6%8E%A7%E5%88%B6

流控模式--直接

nacos中配置规则：

[
       {
        "resource": "/todolist/getTestList",
        "controlBehavior":0,
        "count":2,
        "grade":1,
        "limitApp": "default",
        "strategy":0
    },
     {
        "resource": "todolist/getTestListSentinelFlow",
        "controlBehavior":0,
        "count":2,
        "grade":1,
        "limitApp": "default",
        "strategy":0
    }
]

Dashboard回显规则：

image

流控模式--关联

关联的资源访问达到配置的阈值，则此资源的访问会被流控，关联的资源自己不需要做配置。

业务场景举例：在获取订单的接口/order/get中配置了管理资源为/order/add，并且设置了QPS为2，则当/order/add接口每秒访问大于2次，/order/get接口直接限流；此处接口资源/order/add没有做任何配置。

image

流控模式--链路

资源通过调用关系，形成调用链路，构成一棵调用树。

业务场景举例：资源/order/test1和/order/test2都调用了资源getUser,Sentinel只允许根据某个入口的统计信息做限流。同时调用getUser方法，对/order/test1进行限流，/order/test2不限流。

image image

流控效果

快速失败：默认的流量控制方式，当QPS超过任意规则的阈值后，新的请求就会被立即拒绝，拒绝方式为抛出FlowException。这种方式适用于处理能力已知的情况下，比如通过压测确定了系统的准确水位后。

Warm up（激增流量）：预热/冷启动方式，当系统长期处于低水位的情况下，当流量突然增加时，直接把系统拉升到可能瞬间把系统拉垮。通过“冷启动”，让通过的流量缓慢增加，在一定时间内逐渐增加到阈值上限，给冷系统一个预热的时间，避免冷系统被压垮。

冷加载因子：codeFactor默认是3，即QPS从threshold/3开始，经过预热时长逐渐升至指定的QPS阈值。

image

排队等待（脉冲流量）：严格控制请求通过的间隔时间，也即是让请求以均匀的速度通过。该方式的作用如下图所示：

image

这种方式主要用于处理间隔性突发的流量，例如消息队列。在某一秒有大量的请求进来，而接下来的几秒则处于空闲状态，我们希望系统能在接下来的空闲时间逐渐处理这些请求，而不是在第一秒直接拒绝多余的请求。

注意：暂不支持QPS>1000的排队等待处理

熔断降级

 远程服务不稳定或者网络抖动时暂时关闭，就叫服务熔断。当某个服务熔断之后，服务将不再调用，此时客户端可以自己准备一个本地的fallback（回退）回调，返回一个缺省值。虽然服务水平下降，但好歹可用，总比直接宕机要强，这也要看适合的场景。

image

熔断降级方式，则引入的rule-type为degrade。

nacos中配置规则：

[
  {
  "resource": "todolist/getTestListSentinelDegrade",
  "grade":0,
  "count":1000,
  "timeWindow":3,
  "minRequestAmount": 2,
  "statIntervalMs":3000,
  "slowRatioThreshold":0.1
  }
]

Dashboard回显规则：

image

熔断策略

慢调用比例：慢调用比例作为阈值，需要设置运行慢调用的RT(即最大的响应时间ms），请求的响应时间大于该值则统计为慢调用。当单位统计时长内请求数目大于设置的最小请求数，并且慢调用的比例大于阈值，则接下来的熔断时长内请求会被熔断，经过熔断时长后熔断器会进入探测恢复状态（HALF-OPEN）,若接下来的一个请求响应时间小于设置的慢调用RT则结束熔断，若大于设置的慢调用RT则会再次被熔断。

例子：每次请求处理的最大时长为1秒，大于1秒则为慢调用；3秒内，请求数大于2，且有1次是慢调用，则熔断服务3秒

image

异常比例：单位统计时长内请求数目大于设置的最小请求数，并且异常的比例大于阈值，则接下来的熔断时长内请求自动熔断

异常数：当单位统计时长内的异常数目超过阈值之后会自动进行熔断。

系统自适应保护

 Sentinel 同时提供系统维度的自适应保护能力。防止雪崩，是系统防护中重要的一环。当系统负载较高的时候，如果还持续让请求进入，可能会导致系统崩溃，无法响应。

系统自适应保护，则引入的rule-type为system。

例子：cpu使用率阈值为5%，大于5%则触发系统保护规则。

nacos中配置规则：

[
  {
    "resource": "todolist/system",
    "highestSystemLoad":3.0,
    "highestCpuUsage":0.05,
    "avgRt":10,
    "qps": 20,
    "maxThread":10
  }
]

image image

load自适应（仅对Linux/unix-like机器生效）：系统的load1作为启发指标，进行自适应系统保护。当系统load1超过设定的启发值，且系统当前的并发线程数超过估算的系统容量时，才会触发系统保护（BBR阶段）。

RT:当单台机器上所有入口流量的评价RT达到阈值时触发系统保护，单位是毫秒。

线程数：当单台机器上所有入口流量的并发线程数达到阈值即触发系统保护。

入口QPS：当单台机器上所有入口流量的QPS到阈值即触发系统保护。

CPU使用率：当系统CPU使用率超过阈值即触发系统保护（取值范围0.0-1.0），比较灵活。

热点参数限流

 热点参数限流会统计传入参数中的热点参数，并根据配置的限流阈值与模式，对包含热点参数的资源调用进行限流。热点参数限流可以看做是一种特殊的流量控制，仅对包含热点参数的资源调用生效。

image

热点参数，则引入的rule-type为param-flow。

例：给接口getById的第一个参数（index为0）添加热点限流，普通参数qps最大2，参数为1时，最大qps可以达到5。

nacos中配置规则：

[
  {
    "burstCount": 0,
    "clusterMode": false,
    "controlBehavior": 0,
    "count": 2,
    "durationInSec": 1,
    "grade": 1,
    "limitApp": "default",
    "maxQueueingTimeMs": 0,
    "paramFlowItemList": [
      {
      "classType": "int",
      "count": 5,
      "object": "1"
      }
    ],
    "paramIdx": 0,
    "resource": "todolist/getById"
  }
]

Dashboard回显规则：

image

黑白名单控制

很多时候，我们需要根据调用来源来判断该次请求是否允许放行，这时候可以使用 Sentinel 的来源访问控制（黑白名单控制）的功能。来源访问控制根据资源的请求来源（origin）限制资源是否通过，若配置白名单则只有请求来源位于白名单内时才可通过；若配置黑名单则请求来源位于黑名单时不通过，其余的请求通过。

系统自适应保护，则引入的rule-type为authority。

例：给ip为10.18.xx.xx的请求来源添加对资源/todolist/getTestList1访问的黑名单。

nacos中配置规则：

[
    {
        "resource": "/todolist/getTestList1",
        "limitApp": "10.18.xx.xx",
        "strategy": 1
    }
]

项目工程中添加解析来源ip的配置类：

@Component
public class IpLimitConifg implements RequestOriginParser {

    //解析调用的ip
    @Override
    public String parseOrigin(HttpServletRequest request) {
        return request.getRemoteAddr();
    }
}

Dashboard回显规则：

image

Sentinel实现原理

在 Sentinel 里面，所有的资源都对应一个资源名称（resourceName），每次资源调用都会创建一个 Entry 对象。Entry 可以通过对主流框架的适配自动创建，也可以通过注解的方式或调用 SphU API 显式创建。Entry 创建的时候，同时也会创建一系列功能插槽（slot chain），这些插槽有不同的职责，根据定义的资源名创建的功能插槽chain如下：

image image

• first 链表的头结点；
• NodeSelectorSlot 负责收集资源的路径，并将这些资源的调用路径，以树状结构存储起来，用于根据调用路径来限流降级；
• ClusterBuilderSlot 则用于存储资源的统计信息以及调用者信息，例如该资源的 RT, QPS, thread count 等等，这些信息将用作为多维度限流，降级的依据；
• LogSlot 日志打印；
• StatisticSlot 则用于记录、统计不同纬度的 runtime 指标监控信息；
• AuthoritySlot 则根据配置的黑白名单和调用来源信息，来做黑白名单控制；
• SystemSlot 则通过系统的状态，例如 load1 等，来控制总的入口流量；
• FlowSlot 则用于根据预设的限流规则以及前面 slot 统计的状态，来进行流量控制；
• DegradeSlot 则通过统计信息以及预设的规则，来做熔断降级；
• end 链表的尾结点。

现在以SphU.entry方法为切入点来开始分析。这个方法会去申请一个entry，如果能够申请成功，则说明没有被限流，否则会抛出BlockException，表明已经被限流了。

sentinel的责任链的传递方式：每个Slot节点执行完自己的业务后，会调用fireEntry来触发下一个节点的entry方法，就通过SlotChain完成了对每个节点的entry()方法的调用，每个节点会根据创建的规则，进行自己的逻辑处理，当统计的结果达到设置的阈值时，就会触发限流、降级等事件，具体就是抛出BlockException异常。

源码分析

以SphU.entry(RESOURCE_NAME)进入流控的源码规则创建。

调用的是SphU.java类的entry方法:

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

接下来调用的是Env类中创建的CtSph类，CtSph类实现了Sph接口，CtSph类的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);
    }

根据传递的资源名和EntryType类型创建StringResourceWrapper，再调用CtSph类中的entry方法：

public Entry entry(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
        return entryWithPriority(resourceWrapper, count, false, args);
    }

再调用CtSph类的entryWithPriority方法：

private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
        throws BlockException {
        Context context = ContextUtil.getContext();
        if (context instanceof NullContext) {
            // The {@link NullContext} indicates that the amount of context has exceeded the threshold,
            // so here init the entry only. No rule checking will be done.
            return new CtEntry(resourceWrapper, null, context);
        }

        if (context == null) {
            // Using default context.
            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;
    }

主要关注CtSph类创建责任链的方法lookProcessChain(resourceWrapper)：

 ProcessorSlot<Object> lookProcessChain(ResourceWrapper resourceWrapper) {
        ProcessorSlotChain chain = chainMap.get(resourceWrapper);
        if (chain == null) {
            synchronized (LOCK) {
                chain = chainMap.get(resourceWrapper);
                if (chain == null) {
                    // Entry size limit.
                    if (chainMap.size() >= Constants.MAX_SLOT_CHAIN_SIZE) {
                        return null;
                    }

                    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;
    }

当我们第一次访问资源名为“hello”的时候，从chainMap（全局资源责任链集合）中获取不到此资源的任何定义，然后使用synchronized加锁双校验是否为空的检查方式校验是否为空（防止并发线程不安全），为空则开始创建责任链chain；Constants.MAX_SLOT_CHAIN_SIZE最大值为6000，目前责任链最大值定义是6000。chainMap是一个map集合：

    private static volatile Map<ResourceWrapper, ProcessorSlotChain> chainMap
        = new HashMap<ResourceWrapper, ProcessorSlotChain>();

使用volatile关键字进行修饰，为了线程安全使用缓存一致性协议（MESI），当map中数据变动后其他线程能通过总线嗅探机制感知，然后及时失效自己工作内存中的数据并从主内存中重新获取数据写入自己的工作内存。

从chainMap中获取不到资源的定义（第一次访问），使用SlotChainProvider类创建责任链SlotChainProvider.newSlotChain()：

   public static ProcessorSlotChain newSlotChain() {
        if (slotChainBuilder != null) {
            return slotChainBuilder.build();
        }

        // Resolve the slot chain builder SPI.
        slotChainBuilder = SpiLoader.of(SlotChainBuilder.class).loadFirstInstanceOrDefault();

        if (slotChainBuilder == null) {
            // Should not go through here.
            RecordLog.warn("[SlotChainProvider] Wrong state when resolving slot chain builder, using default");
            slotChainBuilder = new DefaultSlotChainBuilder();
        } else {
            RecordLog.info("[SlotChainProvider] Global slot chain builder resolved: {}",
                slotChainBuilder.getClass().getCanonicalName());
        }
        return slotChainBuilder.build();
    }

第一次的时候，责任链的构造接口类SlotChainBuilder为空，通过类SpiLoader构造一个责任链的构造类：

 slotChainBuilder = SpiLoader.of(SlotChainBuilder.class).loadFirstInstanceOrDefault()

使用责任链的构造类创建插槽，调用slotChainBuilder.build()。

DefaultSlotChainBuilder类实现了SlotChainBuilder接口，所以执行slotChainBuilder.build()的方法在DefaultSlotChainBuilder类中：

  public ProcessorSlotChain build() {
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();

        List<ProcessorSlot> sortedSlotList = SpiLoader.of(ProcessorSlot.class).loadInstanceListSorted();
        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;
    }

DefaultProcessorSlotChain类继承了ProcessorSlotChain类，所以使用DefaultProcessorSlotChain创建类：

public class DefaultProcessorSlotChain extends ProcessorSlotChain {

    AbstractLinkedProcessorSlot<?> first = new AbstractLinkedProcessorSlot<Object>() {

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

        @Override
        public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
            super.fireExit(context, resourceWrapper, count, args);
        }

    };
    AbstractLinkedProcessorSlot<?> end = first;

    @Override
    public void addFirst(AbstractLinkedProcessorSlot<?> protocolProcessor) {
        protocolProcessor.setNext(first.getNext());
        first.setNext(protocolProcessor);
        if (end == first) {
            end = protocolProcessor;
        }
    }

    @Override
    public void addLast(AbstractLinkedProcessorSlot<?> protocolProcessor) {
        end.setNext(protocolProcessor);
        end = protocolProcessor;
    }

    /**
     * Same as {@link #addLast(AbstractLinkedProcessorSlot)}.
     *
     * @param next processor to be added.
     */
    @Override
    public void setNext(AbstractLinkedProcessorSlot<?> next) {
        addLast(next);
    }

    @Override
    public AbstractLinkedProcessorSlot<?> getNext() {
        return first.getNext();
    }

    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args)
        throws Throwable {
        first.transformEntry(context, resourceWrapper, t, count, prioritized, args);
    }

    @Override
    public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
        first.exit(context, resourceWrapper, count, args);
    }

}

先创建出chain的DefaultProcessorSlotChain类，使用属性first存放DefaultProcessorSlotChain类，并把first赋值给end：

AbstractLinkedProcessorSlot <?> end = first;

类SpiLoader加载需要的9个插槽slot：

image

循环把这9个slot添加到Chain责任链中：

     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);
        }

DefaultProcesserSlotChain类添加到责任链最后位置方法：

    public void addLast(AbstractLinkedProcessorSlot<?> protocolProcessor) {
        end.setNext(protocolProcessor);
        end = protocolProcessor;
    }

初始的时候把first赋值给了end，addLast方法操作的是end属性，每次都添加到end属性的下一个节点，并且把当前节点赋值给end属性，循环把这九个插槽赋值到责任链上。

构建出来的责任链添加到全局chainMap资源中：

                    Map<ResourceWrapper, ProcessorSlotChain> newMap = new HashMap<ResourceWrapper, ProcessorSlotChain>(
                        chainMap.size() + 1);
                    newMap.putAll(chainMap);
                    newMap.put(resourceWrapper, chain);
                    chainMap = newMap;

添加的方式为：创建一个临时的map对象，大小为全局chainMap大小加1，先把chainMap添加到临时的map中，再把刚才创建的责任链添加到临时map中，key为根据资源名构建的wrapper对象，value为责任链；再把整个临时map赋值给全局chainMap。

创建完责任链后，根据资源名构建的wrapper和chain和访问的context构建Entry对象。

Entry e = new CtEntry(resourceWrapper, chain, context)

resourceWrapper构建的资源截图：

image

责任链chain截图：

image

context内容截图：

image

通过调用责任链的entry方法，若是调用异常，则抛出BlockException：

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);
}

执行DefaultProcessorSlotChain类的entry方法：

public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args)
    throws Throwable {
    first.transformEntry(context, resourceWrapper, t, count, prioritized, args);
}

接着执行first.transformEntry()，先执行责任链的第一个节点first，first的类型是AbstractLinkedProcessorSlot类，所以执行AbstractLinkedProcessorSlot的transformEntry方法：

void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, int count, boolean prioritized, Object... args)
    throws Throwable {
    T t = (T)o;
    entry(context, resourceWrapper, t, count, prioritized, args);
}

当前的t强转后为null，再调用entry后，调用到AbstractLinkedProcessorSlot的entry方法：

public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args)
    throws Throwable {
    super.fireEntry(context, resourceWrapper, t, count, prioritized, args);
}

再调用AbstractLinkedProcessorSlot的fireEntry方法：

public void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
    throws Throwable {
    if (next != null) {
        next.transformEntry(context, resourceWrapper, obj, count, prioritized, args);
    }
}

调用到AbstractLinkedProcessorSlot的transformEntry方法：

void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, int count, boolean prioritized, Object... args)
    throws Throwable {
    T t = (T)o;
    entry(context, resourceWrapper, t, count, prioritized, args);
}

此entry方法的实现类是NodeSelectSlot：

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);

同样的每个插槽都有一个全局map，用于记录资源的访问情况，map也是使用volatile修饰（保证每个线程拿到的数据一致），同样使用synchronized加锁双非空校验赋值。

执行完后调用fireEntry方法，开始同样的方式创建责任链的next节点，此时调用到AbstractLinkedProcessorSlot类的fireEntry方法，再调用transformEntry方法，接着调用entry进入到具体的T转成的插槽中执行方法。

权限AuthoritySlot插槽会根据设置的黑白名单进行校验，entry方法：

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);
}

checkBlackWhiteAuthority先拿到配置的权限规则，若是没有配置直接通过，有配置再进行校验，校验不通过直接抛出AuthorityException异常，方法：

void checkBlackWhiteAuthority(ResourceWrapper resource, Context context) throws AuthorityException {
    Map<String, Set<AuthorityRule>> authorityRules = AuthorityRuleManager.getAuthorityRules();

    if (authorityRules == null) {
        return;
    }

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

    for (AuthorityRule rule : rules) {
        if (!AuthorityRuleChecker.passCheck(rule, context)) {
            throw new AuthorityException(context.getOrigin(), rule);
        }
    }
}

系统SystemSlot插槽会根据设置的规则校验，entry方法：

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);
}

SystemRuleManager.checkSystem校验系统规则，若是有违背规则的直接抛出SystemBlockException异常：

public static void checkSystem(ResourceWrapper resourceWrapper) throws BlockException {
    if (resourceWrapper == null) {
        return;
    }
    // Ensure the checking switch is on.
    if (!checkSystemStatus.get()) {
        return;
    }

    // for inbound traffic only
    if (resourceWrapper.getEntryType() != EntryType.IN) {
        return;
    }

    // total qps
    double currentQps = Constants.ENTRY_NODE == null ? 0.0 : Constants.ENTRY_NODE.successQps();
    if (currentQps > qps) {
        throw new SystemBlockException(resourceWrapper.getName(), "qps");
    }

    // total thread
    int currentThread = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.curThreadNum();
    if (currentThread > maxThread) {
        throw new SystemBlockException(resourceWrapper.getName(), "thread");
    }

    double rt = Constants.ENTRY_NODE == null ? 0 : Constants.ENTRY_NODE.avgRt();
    if (rt > maxRt) {
        throw new SystemBlockException(resourceWrapper.getName(), "rt");
    }

    // load. BBR algorithm.
    if (highestSystemLoadIsSet && getCurrentSystemAvgLoad() > highestSystemLoad) {
        if (!checkBbr(currentThread)) {
            throw new SystemBlockException(resourceWrapper.getName(), "load");
        }
    }

    // cpu usage
    if (highestCpuUsageIsSet && getCurrentCpuUsage() > highestCpuUsage) {
        throw new SystemBlockException(resourceWrapper.getName(), "cpu");
    }
}

热点参数ParamFlowSlot插槽entry方法，没有配置热点参数直接调用fireEntry执行下一个：

public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
                  boolean prioritized, Object... args) throws Throwable {
    if (!ParamFlowRuleManager.hasRules(resourceWrapper.getName())) {
        fireEntry(context, resourceWrapper, node, count, prioritized, args);
        return;
    }

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

校验热点参数，有异常抛出ParamFlowException异常：

void checkFlow(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
    if (args == null) {
        return;
    }
    if (!ParamFlowRuleManager.hasRules(resourceWrapper.getName())) {
        return;
    }
    List<ParamFlowRule> rules = ParamFlowRuleManager.getRulesOfResource(resourceWrapper.getName());

    for (ParamFlowRule rule : rules) {
        applyRealParamIdx(rule, args.length);

        // Initialize the parameter metrics.
        ParameterMetricStorage.initParamMetricsFor(resourceWrapper, rule);

        if (!ParamFlowChecker.passCheck(resourceWrapper, rule, count, args)) {
            String triggeredParam = "";
            if (args.length > rule.getParamIdx()) {
                Object value = args[rule.getParamIdx()];
                triggeredParam = String.valueOf(value);
            }
            throw new ParamFlowException(resourceWrapper.getName(), triggeredParam, rule);
        }
    }
}

流控FlowSlot插槽entry方法：

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);
}

校验流控规则，有异常抛出FlowException异常：

public void checkFlow(Function<String, Collection<FlowRule>> ruleProvider, ResourceWrapper resource,
                      Context context, DefaultNode node, int count, boolean prioritized) throws BlockException {
    if (ruleProvider == null || resource == null) {
        return;
    }
    Collection<FlowRule> rules = ruleProvider.apply(resource.getName());
    if (rules != null) {
        for (FlowRule rule : rules) {
            if (!canPassCheck(rule, context, node, count, prioritized)) {
                throw new FlowException(rule.getLimitApp(), rule);
            }
        }
    }
}

获取到流控规则的配置项信息：

image

降级DegradeSlot插槽entry方法：

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

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

降级规则校验，有异常抛出DegradeException：

void performChecking(Context context, ResourceWrapper r) throws BlockException {
    List<CircuitBreaker> circuitBreakers = DegradeRuleManager.getCircuitBreakers(r.getName());
    if (circuitBreakers == null || circuitBreakers.isEmpty()) {
        return;
    }
    for (CircuitBreaker cb : circuitBreakers) {
        if (!cb.tryPass(context)) {
            throw new DegradeException(cb.getRule().getLimitApp(), cb.getRule());
        }
    }
}

校验流程图：

image