深入理解Sentinel：11 黑白名单限流与热点参数限流

黑白名单限流黑白名单过滤是使用最为广泛的一种过滤规则例如用于实现接口安全的 IP 黑白名单规则过滤用于防骚扰的短信、来电拦截黑白名单过滤。所以 Sentinel 中的黑白名单限流并不难理解如果配置了黑名单且请求来源存在黑名单中则拦截拒绝请求如果配置了白名单且请求来源存在白名单中则放行。Sentinel 不支持一个黑白名单规则同时配置黑名单和白名单因此不存优先级的问题。黑白名单过滤功能更像是一种授权机制它简单的将权限分为有权限和无权限两种情况如果支持冲突可使用优先级策略解决冲突问题。Sentinel 把黑白名作为授权策略实现黑白名单限流即实现授权限流。Sentinel 在命名上也是使用 Authority而非 BlackWhiteList。一些关键类说明AuthoritySlot实现黑白名称授权功能的切入点ProcessorSlotAuthorityRule授权规则类AuthorityRuleChecker授权检测类AuthorityRuleManager授权规则管理者提供 loadRuls APIAuthorityException授权检测异常继承 BlockExceptionAuthorityRule授权规则AuthorityRule是 Sentinel 中最易于理解的一种规则AuthorityRule 的配置项如下public class AuthorityRule extends AbstractRule { private int strategy RuleConstant.AUTHORITY_WHITE; }resource资源名称从父类继承而来。limitApp限制的来源名称在 AuthorityRule 中可配置多个使用‘,’号分隔。strategy限流策略白名单AUTHORITY_WHITE黑名单AUTHORITY_BLACK。当 strategy 配置为 AUTHORITY_WHITE 时limitApp 即为白名单当 strategy 配置为 AUTHORITY_BLACK 时limitApp 即为黑明单。例如AuthorityRule rule new AuthorityRule(); // 资源名称 rule.setResource(GET:/hello); // 白名单策略 rule.setStrategy(RuleConstant.AUTHORITY_WHITE); // 白名单 rule.setLimitApp(serviceA,serviceC); AuthorityRuleManager.loadRules(Collections.singletonList(rule));上述规则用于限制资源 “GET:/hello” 只允许服务 A 和服务 C 访问。AuthoritySlot在使用默认的 SlotChainBuilder 情况下AuthoritySlot 被放在 SystemSlot、FlowSlot、DegradeSlot 的前面其优先级更高。原因之一是授权限流不需要使用统计的指标数据另一个原因则是提升性能在未授权的情况下没必要判断是否需要熔断、系统负载能否接住这个请求、QPS 是否过高等这与用户授权功能是一样的道理未登陆无需判断是否有权限访问某个资源。AuthoritySlot 的实现源码如下public class AuthoritySlot extends AbstractLinkedProcessorSlot { 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); } Override public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { fireExit(context, resourceWrapper, count, args); } void checkBlackWhiteAuthority(ResourceWrapper resource, Context context) throws AuthorityException { // (1) Map authorityRules AuthorityRuleManager.getAuthorityRules(); if (authorityRules null) { return; } // (2) Set rules authorityRules.get(resource.getName()); if (rules null) { return; } // (3) for (AuthorityRule rule : rules) { if (!AuthorityRuleChecker.passCheck(rule, context)) { throw new AuthorityException(context.getOrigin(), rule); } } } }从 AuthorityRuleManager 获取当前配置的所有授权规则获取为当前资源配置的所有授权规则遍历授权规则调用 AuthorityRuleChecker#passCheck 方法判断是否拒绝当前请求是则抛出 AuthorityException 异常。AuthorityRuleCheckerAuthorityRuleChecker 负责实现黑白名单的过滤逻辑其 passCheck 方法源码如下static boolean passCheck(AuthorityRule rule, Context context) { // 获取来源 String requester context.getOrigin(); // 来源为空或者来源等于规则配置的 limitApp 则拦截请求 if (StringUtil.isEmpty(requester) || StringUtil.isEmpty(rule.getLimitApp())) { return true; } // 字符串查找这一步起到快速过滤的作用提升性能 int pos rule.getLimitApp().indexOf(requester); boolean contain pos -1; // 存在才精确匹配 if (contain) { boolean exactlyMatch false; // 分隔数组 String[] appArray rule.getLimitApp().split(,); for (String app : appArray) { if (requester.equals(app)) { // 标志设置为 true exactlyMatch true; break; } } contain exactlyMatch; } // 策略 int strategy rule.getStrategy(); // 如果是黑名单且来源存在规则配置的黑名单中 if (strategy RuleConstant.AUTHORITY_BLACK contain) { return false; } // 如果是白名单且来源不存在规则配置的白名单中 if (strategy RuleConstant.AUTHORITY_WHITE !contain) { return false; } return true; }整个方法都比较简单首先是从当前 Context 获取调用来源的名称只有在调用来源不为空且规则配置了黑名单或者白名单的情况下才会走黑白名单的过滤逻辑这也说明要实现黑白名单限流的前提是每个服务消费端在发起请求时都必须要携带自身服务的名称这取决于 Sentinel 主流框架适配器其次Sentinel 通过使用 indexOf 先简单匹配一次黑名单或白名单再切割黑名单或白名单数组实现精确匹配这有助于提升性能如果当前请求来源存在名单中则根据策略判断这份名称是黑名单还是白名单再决定是否需要拒绝请求。热点参数限流热点参数限流并非在 Sentinel 的 core 模块中实现的但也是非常实用的一种限流方式。并且Sentinel 支持 API Gateway 网关限流也是基于参数限流实现的了解热点参数限流的实现原理也有助于我们更好地理解网关限流。参数限流即根据方法调用传递的参数实现限流又或者说是根据接口的请求参数限流热点参数限流即针对访问频繁的参数限流。例如都是调用一个下单接口但购买的商品不同比如主播带货的商品下单流量较大而一般商品购买量很少同时因为商品数量有限不太可能每个下单请求都能购买成功如果能实现根据客户端请求传递的商品 ID 实现限流将流量控制在商品的库存总量左右并且使用 QPS 限流等兜底这种有针对性的限流将接口通过的有效流量最大化。热点参数限流功能在 Sentinel 源码的扩展功能模块为 sentinel-extension子模块为 sentinel-parameter-flow-control。基于滑动窗口的热点参数指标数据统计热点参数限流使用的指标数据不再是 core 模块中统计的指标数据而是重新实现了一套指标数据统计功能依旧是基于滑动窗口。ParamMapBucket实现参数指标数据统计的 Bucket用于统计某个参数对应不同取值的被限流总数、被放行的总数。HotParameterLeapArray实现滑动窗口持有 WindowWrap 数组WindowWrap 包装 ParamMapBucket。与 core 模块的 MetricBucket 实现不同MetricBucket 只统计每个指标的数值而 ParamMapBucket 需要统计每个指标、参数的每种取值的数值MetricBucket 更像是 Redis 中的 String 结构而 ParamMapBucket 更像 Redis 中的 Hash 结构。ParamMapBucket 的源码如下public class ParamMapBucket { // 数组类型为 CacheMap private final CacheMap[] data; public ParamMapBucket() { this(DEFAULT_MAX_CAPACITY); } public ParamMapBucket(int capacity) { RollingParamEvent[] events RollingParamEvent.values(); // 根据需要统计的指标数据创建数组 this.data new CacheMap[events.length]; // RollingParamEvent 可取值为 REQUEST_PASSED、REQUEST_BLOCKED for (RollingParamEvent event : events) { data[event.ordinal()] new ConcurrentLinkedHashMapWrapper(capacity); } } }data数组元素类型为CacheMapObject, AtomicInteger下标为 0 存储的是统计请求通过的指标数据下标为 1 统计的是请求被拒绝的指标数据。CacheMapObject, AtomicIntegerkey 为参数的取值例如商品的 IDvalue 才是指标数值。HotParameterLeapArray 继承 LeapArray即实现滑动窗口。ParamMapBucket 不存储窗口时间信息窗口时间信息依然由 WindowWrap 存储HotParameterLeapArray 使用 WindowWrap 包装 ParamMapBucket。笔者也是看了 HotParameterLeapArray 之后才明白为什么 Sentienl 将滑动窗口抽象为 LeapArray这为扩展实现收集自定义指标数据的滑动窗口提供了支持。HotParameterLeapArray 的提供的几个 API 如下public class HotParameterLeapArray extends LeapArray { //..... public void addValue(RollingParamEvent event, int count, Object value) { // .... } public Map getTopValues(RollingParamEvent event, int number) { // ..... } public long getRollingSum(RollingParamEvent event, Object value) { // ..... } public double getRollingAvg(RollingParamEvent event, Object value) { // .... } }addValue添加参数的指标数值例如给 REQUEST_PASSED 指标且参数取值为 4343433 的指标数值加上 count假设这个滑动窗口是用于统计商品 ID 参数的4343433 表示商品 IDcount 为 1调用该方法表示给商品 ID 为 4343433 的请求通过总数加 1。getTopValues获取热点参数的 QPS即获取某个指标排名前 number 的参数取值与指标数据。例如获取 REQUEST_PASSED 指标排名前 10 的 QPS方法返回值类型为 Mapkey 为参数的取值value 为 QPS。getRollingSum计算某个指标、参数的某个取值的总请求数。例如获取 REQUEST_PASSED 且商品 ID 为 4343433 的请求总数。getRollingAvg获取某个指标、参数的某个取值的平均 QPS。例如获取 REQUEST_PASSED 且商品 ID 为 4343433 的平均 QPS。可见如果是分钟级的滑动窗口一分内参数的取值越多其占用的内存就越多。参数限流中的 Node两个需要重点关注的类ParameterMetric用于实现类似 ClusterNode 的统计功能。ParameterMetricStorage用于实现类似 EntranceNode 功能管理和存储每个资源对应的 ParameterMetric。ParameterMetric 有三个静态字段源码如下public class ParameterMetric { private final Map ruleTimeCounters new HashMap(); private final Map ruleTokenCounter new HashMap(); private final Map threadCountMap new HashMap(); }ruleTimeCounters用于实现匀速流量控制效果key 为参数限流规则ParamFlowRule值为参数不同取值对应的上次生产令牌的时间。ruleTokenCounter用于实现匀速流量控制效果key 为参数限流规则ParamFlowRule值为参数不同取值对应的当前令牌桶中的令牌数。threadCountMapkey 为参数索引值为参数不同取值对应的当前并行占用的线程总数。ParameterMetricStorage 使用 ConcurrentHashMap 缓存每个资源对应的 ParameterMetric只会为配置了参数限流规则的资源创建 ParameterMetric。其部份源码如下所示public final class ParameterMetricStorage { private static final Map metricsMap new ConcurrentHashMap(); private static final Object LOCK new Object(); public static void initParamMetricsFor(ResourceWrapper resourceWrapper,ParamFlowRule rule) { if (resourceWrapper null || resourceWrapper.getName() null) { return; } String resourceName resourceWrapper.getName(); ParameterMetric metric; // 双重检测线程安全为资源创建全局唯一的 ParameterMetric if ((metric metricsMap.get(resourceName)) null) { synchronized (LOCK) { if ((metric metricsMap.get(resourceName)) null) { metric new ParameterMetric(); metricsMap.put(resourceWrapper.getName(), metric); } } } // 初始化 ParameterMetric metric.initialize(rule); } }initParamMetricsFor 方法用于为资源创建 ParameterMetric 并初始化该方法在资源被访问时由 ParamFlowSlot 调用并且该方法只在为资源配置了参数限流规则的情况下被调用。热点参数限流功能的实现sentinel-parameter-flow-control 模块通过 Java SPI 注册自定义的 SlotChainBuilder即注册 HotParamSlotChainBuilder将 ParamFlowSlot 放置在 StatisticSlot 的后面这个 ParamFlowSlot 就是实现热点参数限流功能的切入点。public class ParamFlowSlot extends AbstractLinkedProcessorSlot { Override 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); } Override public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) { fireExit(context, resourceWrapper, count, args); } }既然是参数限流那么肯定是需要能够拿到参数了而 ProcessorSlot#entry 方法的最后一个参数就是请求传递过来的参数通过 SphU#entry 方法一层层往下传递。例如GetMapping(/hello) public String apiHello(String name) throws BlockException { ContextUtil.enter(my_context); Entry entry null; try { entry SphU.entry(GET:/hello, EntryType.IN,1,name); doBusiness(); return Hello!; } catch (Exception e) { if (!(e instanceof BlockException)) { Tracer.trace(e); } throw e; } finally { if (entry ! null) { entry.exit(1); } ContextUtil.exit(); } }当 SphU#entry 调用到 ParamFlowSlot#entry 方法时ParamFlowSlot 调用 checkFlow 方法判断是否需要限流。checkFlow 方法的实现如下void checkFlow(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException { //1 if (args null) { return; } if (!ParamFlowRuleManager.hasRules(resourceWrapper.getName())) { return; } List rules ParamFlowRuleManager.getRulesOfResource(resourceWrapper.getName()); //2 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); } } }checkFlow 方法的最后一个参数是请求参数也就是调用 SphU#entry 方法传递进来的参数。checkFlow 方法首先调用 ParamFlowRuleManager 的 API 判断当前资源有没有配置参数限流规则如果有则获取为当前资源配置的所有参数限流规则。遍历参数限流规则调用 ParameterMetricStorage#initParamMetricsFor 方法判断是否需要为当前资源初始化创建 ParameterMetric然后调用 ParamFlowChecker#passCheck 方法判断当前请求是否可以放行如果需要拒绝请求则抛出 ParamFlowException 异常。在阅读 ParamFlowChecker#passCheck 方法的源码之前我们需要先了解参数限流规则的配置了解每个配置项的作用。参数限流规则 ParamFlowRule 的源码如下有删减public class ParamFlowRule extends AbstractRule { private int grade RuleConstant.FLOW_GRADE_QPS; private double count; private Integer paramIdx; private int controlBehavior RuleConstant.CONTROL_BEHAVIOR_DEFAULT; private int maxQueueingTimeMs 0; private long durationInSec 1; private int burstCount 0; }grade限流规则的阈值类型支持的类型同 FlowRule。count阈值同 FlowRule。paramIdx参数索引ParamFlowChecker 根据限流规则的参数索引获取参数的值下标从 0 开始例如方法public String apiHello(String name)该方法只有一个参数索引为 0 对应 name 参数。controlBehavior流量控制效果同 FlowRule但只支持快速失败和匀速排队。maxQueueingTimeMs实现匀速排队流量控制效果的虚拟队列最大等待时间超过该值的请求被抛弃同 FlowRuledurationInSec统计指标数据的窗口时间大小单位为秒。burstCount支持的突发流量总数。假设需要针对资源“GET:/hello”的 name 参数限流当 name 取值为“jackson”时限流 QPS 阈值为 5则配置如下ParamFlowRule rule new ParamFlowRule(); // 资源为/hello rule.setResource(GET:/hello); // 索引 0 对应的参数为 name rule.setParamIdx(0); // qps 限流 rule.setGrade(RuleConstant.FLOW_GRADE_QPS); // 阈值为 5 rule.setCount(5); ParamFlowRuleManager.loadRules(Collections.singletonList(rule));以此为例我们分析 ParamFlowChecker#passCheck 方法源码passCheck 返回 true 表示放行返回 false 表示拒绝。ParamFlowChecker#passCheck 方法源码如下publicstaticbooleanpassCheck(ResourceWrapperresourceWrapper,ParamFlowRulerule,intcount,Object...args){if(argsnull){returntrue;}// 判断参数索引是否合法intparamIdxrule.getParamIdx();if(args.lengthparamIdx){returntrue;}// 获取参数值如果值为空则允许通过Objectvalueargs[paramIdx];if(valuenull){returntrue;}// 集群限流if(rule.isClusterMode()rule.getGrade()RuleConstant.FLOW_GRADE_QPS){returnpassClusterCheck(resourceWrapper,rule,count,value);}// 单机限流returnpassLocalCheck(resourceWrapper,rule,count,value);}如果参数为空、或者参数的总数小于等于规则配置的参数索引值、或者参数索引对应的参数的值为空则放行请求如果是集群限流模式则调用 passClusterCheck 方法否则调用 passLocalCheck 方法。我们先不讨论集群限流情况仅看单机本地限流情况。passLocalCheck 方法的源码如下privatestaticbooleanpassLocalCheck(ResourceWrapperresourceWrapper,ParamFlowRulerule,intcount,Objectvalue){try{// 基本数据类型if(Collection.class.isAssignableFrom(value.getClass())){for(Objectparam:((Collection)value)){if(!passSingleValueCheck(resourceWrapper,rule,count,param)){returnfalse;}}}// 数组类elseif(value.getClass().isArray()){intlengthArray.getLength(value);for(inti0;ilength;i){ObjectparamArray.get(value,i);if(!passSingleValueCheck(resourceWrapper,rule,count,param)){returnfalse;}}}// 引用类型else{returnpassSingleValueCheck(resourceWrapper,rule,count,value);}}catch(Throwablee){}returntrue;}由于参数可能是基本数据类型也可能是数组类型或者引用类型所以 passLocalCheck 方法分三种情况处理。我们只讨论其中一种情况其它情况的实现类似。以资源“GET:/hello”为例其方法 apiHello 的 name 参数为 String 类型因此会调用 passSingleValueCheck 方法该方法源码如下staticbooleanpassSingleValueCheck(ResourceWrapperresourceWrapper,ParamFlowRulerule,intacquireCount,Objectvalue){//1if(rule.getGrade()RuleConstant.FLOW_GRADE_QPS){if(rule.getControlBehavior()RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER){returnpassThrottleLocalCheck(resourceWrapper,rule,acquireCount,value);}else{returnpassDefaultLocalCheck(resourceWrapper,rule,acquireCount,value);}}// 2elseif(rule.getGrade()RuleConstant.FLOW_GRADE_THREAD){SetObjectexclusionItemsrule.getParsedHotItems().keySet();longthreadCountgetParameterMetric(resourceWrapper).getThreadCount(rule.getParamIdx(),value);if(exclusionItems.contains(value)){intitemThresholdrule.getParsedHotItems().get(value);returnthreadCountitemThreshold;}longthreshold(long)rule.getCount();returnthreadCountthreshold;}returntrue;}当规则配置的阈值类型为 QPS 时根据流控效果调用 passThrottleLocalCheck 或 passDefaultLocalCheck 方法当规则配置的阈值类型为 THREAD 时获取当前资源的 ParameterMetric从而取得当前资源、当前参数的值对应的并行占用的线程总 数如果并行占用的线程总数1 大于限流阈值则限流否则放行。你可能好奇并行占用线程总数是在哪里自增和自减的呢这是由 ParamFlowStatisticEntryCallback 与 ParamFlowStatisticExitCallback 这两个 Callback 实现的分别在 StatisticSlot 的 entry 方法和 exit 方法中被回调执行这是我们前面分析 StatisticSlot 源码时故意遗漏的细节。快速失败直接拒绝与匀速排队** 1. 快速失败**快速失败基于令牌桶算法实现。passDefaultLocalCheck 方法控制每个时间窗口只生产一次令牌将令牌放入令牌桶每个请求都从令牌桶中取走令牌当令牌足够时放行当令牌不足时直接拒绝。ParameterMetric#tokenCounters 用作令牌桶timeCounters 存储最近一次生产令牌的时间。passDefaultLocalCheck 方法源码如下根据资源获取 ParameterMetric从 ParameterMetric 获取当前限流规则的令牌桶和最近一次生产令牌的时间时间精确到毫秒。计算限流阈值即令牌桶最大存放的令牌总数tokenCount。重新计算限流阈值将当前限流阈值加上允许突增流量的数量。获取当前时间如果当前参数值未生产过令牌则初始化生产令牌并立即使用maxCount - acquireCount。获取当前时间与上次生产令牌的时间间隔如果间隔时间大于一个窗口时间见6否则见7。计算需要生产的令牌总数并与当前桶中剩余的令牌数相加得到新的令牌总数如果新的令牌总数大于限流阈值则使用限流阈值作为新的令牌总数并且生产完成立即使用maxCount - acquireCount最后更新最近一次生产令牌的时间。从令牌桶中获取令牌如果获取成功oldQpsValue - acquireCount 0则放行当前请求否则拒绝当前请求。2. 匀速排队与 RateLimiterController 实现原理一样passThrottleLocalCheck 方法让请求在虚拟队列中排队控制请求通过的时间间隔该时间间隔通过阈值与窗口时间大小计算出来如果当前请求计算出来的排队等待时间大于限流规则指定的 maxQueueingTimeMs则拒绝当前请求。passThrottleLocalCheck 方法源码如下staticbooleanpassThrottleLocalCheck(ResourceWrapperresourceWrapper,ParamFlowRulerule,intacquireCount,Objectvalue){//1ParameterMetricmetricgetParameterMetric(resourceWrapper);CacheMaptimeRecorderMapmetricnull?null:metric.getRuleTimeCounter(rule);if(timeRecorderMapnull){returntrue;}// 2SetexclusionItemsrule.getParsedHotItems().keySet();longtokenCount(long)rule.getCount();if(exclusionItems.contains(value)){tokenCountrule.getParsedHotItems().get(value);}if(tokenCount0){returnfalse;}//3longcostTimeMath.round(1.0*1000*acquireCount*rule.getDurationInSec()/tokenCount);while(true){//4longcurrentTimeTimeUtil.currentTimeMillis();AtomicLongtimeRecordertimeRecorderMap.putIfAbsent(value,newAtomicLong(currentTime));if(timeRecordernull){returntrue;}longlastPassTimetimeRecorder.get();// 计算当前请求的期望通过时间最近一次请求的期望通过时间 请求通过的时间间隔longexpectedTimelastPassTimecostTime;//5if(expectedTime0){lastPastTimeRef.set(expectedTime);try{TimeUnit.MILLISECONDS.sleep(waitTime);}catch(InterruptedExceptione){RecordLog.warn(passThrottleLocalCheck: wait interrupted,e);}}returntrue;}else{Thread.yield();}}else{returnfalse;}}}当流控效果选择匀速限流时ParameterMetric 的 ruleTimeCounters 不再是记录上次生产令牌的时间而是记录最后一个请求的期望通过时间。计算限流阈值不支持突增流量。计算请求通过的时间间隔例如当 acquireCount 等于 1、限流阈值配置为 200 且窗口时间大小为 1 秒时计算出来的 costTime 等于 5ms即每 5ms 只允许通过一个请求。计算当前请求的期望通过时间值为最近一次请求的期望通过时间 请求通过的时间间隔最近一次请求的期望通过时间也就是虚拟队列中队列尾部的那个请求的期望通过时间。如果期望通过时间与当前时间间隔大于规则配置的允许队列最大等待时间maxQueueingTimeMs则拒绝当前请求否则将当前请求“放入”虚拟队列等待计算出当前请求需要等待的时间让当前线程休眠指定时长之后再放行该请求。总结黑白名单限流的实现相对简单热点参数限流的实现相对复杂。热点参数限流自己实现了一个滑动窗口用于收集指标数据但该滑动窗口并未被使用而是使用 ParameterMetric 与 ParameterMetricStorage这应该是出于性能的考虑。热点参数限流对性能的影响和对内存的占用与参数的取值有多少种可能成正比限流参数的取值可能性越多占用的内存就越大对性能的影响也就越大在使用热点参数限流功能时一定要考虑参数的取值。例如根据商品 ID 限流如果有十万个商品下单那么 CacheMap 将会存在十万个 key-value并且不会被移除随着进程运行的时长而增长。如果限流阈值类型选择为 THREAD 则不会存在这个问题因为在 ParamFlowStatisticExitCallback 方法会调用 ParameterMetric#decreaseThreadCount 方法扣减参数值占用的线程数当线程数为零时会将当前参数值对应的 key-value 从 CacheMap 中移除。