Sentinel源码—7.参数限流和注解的实现

大纲
1.参数限流的原理和源码
2.@SentinelResource注解的使用和实现
 
1.参数限流的原理和源码
(1)参数限流规则ParamFlowRule的配置Demo
(2)ParamFlowSlot根据参数限流规则验证请求
 
(1)参数限流规则ParamFlowRule的配置Demo
一.参数限流的应用场景
二.参数限流规则的属性
三.参数限流规则的配置Demo
 
一.参数限流的应用场景
传统的流量控制，一般是通过资源维度来限制某接口或方法的调用频率。但有时需要更细粒度地控制不同参数条件下的访问速率，即参数限流。参数限流允许根据不同的参数条件设置不同的流量控制规则，这种方式非常适合处理特定条件下的请求，因为能更加精细地管理流量。
 
假设有一个在线电影订票系统，某个接口允许用户查询电影的放映时间。但只希望每个用户每10秒只能查询接口1次，以避免过多的查询请求。这时如果直接将接口的QPS限制为5是不能满足要求的，因为需求是每个用户每5分钟只能查询1次，而不是每秒一共只能查询5次，因此参数限流就能派上用场了。
 
可以设置一个规则，根据用户ID来限制每个用户的查询频率，将限流的维度从资源维度细化到参数维度，从而实现每个用户每10秒只能查询接口1次。比如希望影院工作人员可以每秒查询10次，老板可以每秒查询100次，而购票者则只能每10秒查询一次，其中工作人员的userId值为100和200，老板的userId值为9999，那么可以如下配置：需要注意限流阈值是以秒为单位的，所以需要乘以统计窗口时长10。

二.参数限流规则的属性

1. public class ParamFlowRule extends AbstractRule {
2.     ...
3.     //The threshold type of flow control (0: thread count, 1: QPS).
4.     //流量控制的阈值类型(0表示线程数，1表示QPS)
5.     private int grade = RuleConstant.FLOW_GRADE_QPS;
7.     //Parameter index.
8.     //参数下标
9.     private Integer paramIdx;
11.     //The threshold count.
12.     //阈值
13.     private double count;
15.     //Original exclusion items of parameters.
16.     //针对特定参数的流量控制规则列表
17.     private List<ParamFlowItem> paramFlowItemList = new ArrayList<ParamFlowItem>();
18.    
19.     //Indicating whether the rule is for cluster mode.
20.     //是否集群
21.     private boolean clusterMode = false;
22.     ...
23. }
25. //针对特定参数的流量控制规则
26. public class ParamFlowItem {
27.     private String object;
28.     private Integer count;
29.     private String classType;
30.     ...
31. }

复制代码

三.参数限流规则的配置Demo

1. //This demo demonstrates flow control by frequent ("hot spot") parameters.
2. public class ParamFlowQpsDemo {
3.     private static final int PARAM_A = 1;
4.     private static final int PARAM_B = 2;
5.     private static final int PARAM_C = 3;
6.     private static final int PARAM_D = 4;
8.     //Here we prepare different parameters to validate flow control by parameters.
9.     private static final Integer[] PARAMS = new Integer[] {PARAM_A, PARAM_B, PARAM_C, PARAM_D};
10.     private static final String RESOURCE_KEY = "resA";
11.    
12.     public static void main(String[] args) throws Exception {
13.         initParamFlowRules();
14.         final int threadCount = 20;
15.         ParamFlowQpsRunner<Integer> runner = new ParamFlowQpsRunner<>(PARAMS, RESOURCE_KEY, threadCount, 120);
16.         runner.tick();
17.         Thread.sleep(1000);
18.         runner.simulateTraffic();
19.     }
21.     private static void initParamFlowRules() {
22.         //QPS mode, threshold is 5 for every frequent "hot spot" parameter in index 0 (the first arg).
23.         ParamFlowRule rule = new ParamFlowRule(RESOURCE_KEY)
24.             .setParamIdx(0)
25.             .setGrade(RuleConstant.FLOW_GRADE_QPS)
26.             .setCount(5);
27.         //We can set threshold count for specific parameter value individually.
28.         //Here we add an exception item. That means:
29.         //QPS threshold of entries with parameter `PARAM_B` (type: int) in index 0 will be 10, rather than the global threshold (5).
30.         ParamFlowItem item = new ParamFlowItem().setObject(String.valueOf(PARAM_B))
31.             .setClassType(int.class.getName())
32.             .setCount(10);
33.         rule.setParamFlowItemList(Collections.singletonList(item));
35.         //ParamFlowRuleManager类加载的一个时机是：它的静态方法被调用了
36.         //所以下面会先初始化ParamFlowRuleManager，再执行loadRules()方法
37.         ParamFlowRuleManager.loadRules(Collections.singletonList(rule));
38.     }
39. }
41. public final class ParamFlowRuleManager {
42.     private static final Map<String, List<ParamFlowRule>> PARAM_FLOW_RULES = new ConcurrentHashMap<>();
43.     private final static RulePropertyListener PROPERTY_LISTENER = new RulePropertyListener();
44.     private static SentinelProperty<List<ParamFlowRule>> currentProperty = new DynamicSentinelProperty<>();
45.     static {
46.         currentProperty.addListener(PROPERTY_LISTENER);
47.     }
49.     //Load parameter flow rules. Former rules will be replaced.
50.     public static void loadRules(List<ParamFlowRule> rules) {
51.         try {
52.             //设置规则的值为rules
53.             currentProperty.updateValue(rules);
54.         } catch (Throwable e) {
55.             RecordLog.info("[ParamFlowRuleManager] Failed to load rules", e);
56.         }
57.     }
59.     static class RulePropertyListener implements PropertyListener<List<ParamFlowRule>> {
60.         @Override
61.         public void configUpdate(List<ParamFlowRule> list) {
62.             Map<String, List<ParamFlowRule>> rules = aggregateAndPrepareParamRules(list);
63.             if (rules != null) {
64.                 PARAM_FLOW_RULES.clear();
65.                 PARAM_FLOW_RULES.putAll(rules);
66.             }
67.             RecordLog.info("[ParamFlowRuleManager] Parameter flow rules received: {}", PARAM_FLOW_RULES);
68.         }
70.         @Override
71.         public void configLoad(List<ParamFlowRule> list) {
72.             Map<String, List<ParamFlowRule>> rules = aggregateAndPrepareParamRules(list);
73.             if (rules != null) {
74.                 PARAM_FLOW_RULES.clear();
75.                 PARAM_FLOW_RULES.putAll(rules);
76.             }
77.             RecordLog.info("[ParamFlowRuleManager] Parameter flow rules received: {}", PARAM_FLOW_RULES);
78.         }
79.         ...
80.     }
81.     ...
82. }
84. public class DynamicSentinelProperty<T> implements SentinelProperty<T> {
85.     protected Set<PropertyListener<T>> listeners = new CopyOnWriteArraySet<>();
86.     private T value = null;
87.     public DynamicSentinelProperty() {
89.     }
91.     //添加监听器到集合
92.     @Override
93.     public void addListener(PropertyListener<T> listener) {
94.         listeners.add(listener);
95.         //回调监听器的configLoad()方法初始化规则配置
96.         listener.configLoad(value);
97.     }
99.     //更新值
100.     @Override
101.     public boolean updateValue(T newValue) {
102.         //如果值没变化，直接返回
103.         if (isEqual(value, newValue)) {
104.             return false;
105.         }
106.         RecordLog.info("[DynamicSentinelProperty] Config will be updated to: {}", newValue);
107.         //如果值发生了变化，则遍历监听器，回调监听器的configUpdate()方法更新对应的值
108.         value = newValue;
109.         for (PropertyListener<T> listener : listeners) {
110.             listener.configUpdate(newValue);
111.         }
112.         return true;
113.     }
114.     ...
115. }
117. //A traffic runner to simulate flow for different parameters.
118. class ParamFlowQpsRunner<T> {
119.     private final T[] params;
120.     private final String resourceName;
121.     private int seconds;
122.     private final int threadCount;
123.     private final Map<T, AtomicLong> passCountMap = new ConcurrentHashMap<>();
124.     private final Map<T, AtomicLong> blockCountMap = new ConcurrentHashMap<>();
125.     private volatile boolean stop = false;
127.     public ParamFlowQpsRunner(T[] params, String resourceName, int threadCount, int seconds) {
128.         this.params = params;
129.         this.resourceName = resourceName;
130.         this.seconds = seconds;
131.         this.threadCount = threadCount;
132.         for (T param : params) {
133.             passCountMap.putIfAbsent(param, new AtomicLong());
134.             blockCountMap.putIfAbsent(param, new AtomicLong());
135.         }
136.     }
138.     public void tick() {
139.         Thread timer = new Thread(new TimerTask());
140.         timer.setName("sentinel-timer-task");
141.         timer.start();
142.     }
144.     public void simulateTraffic() {
145.         for (int i = 0; i < threadCount; i++) {
146.             Thread t = new Thread(new RunTask());
147.             t.setName("sentinel-simulate-traffic-task-" + i);
148.             t.start();
149.         }
150.     }
152.     final class TimerTask implements Runnable {
153.         @Override
154.         public void run() {
155.             long start = System.currentTimeMillis();
156.             System.out.println("Begin to run! Go go go!");
157.             System.out.println("See corresponding metrics.log for accurate statistic data");
158.             Map<T, Long> map = new HashMap<>(params.length);
159.             for (T param : params) {
160.                 map.putIfAbsent(param, 0L);
161.             }
162.             while (!stop) {
163.                 sleep(1000);
164.                 //There may be a mismatch for time window of internal sliding window.
165.                 //See corresponding `metrics.log` for accurate statistic log.
166.                 for (T param : params) {
167.                     System.out.println(String.format(
168.                         "[%d][%d] Parameter flow metrics for resource %s: pass count for param <%s> is %d, block count: %d",
169.                         seconds, TimeUtil.currentTimeMillis(), resourceName, param,
170.                         passCountMap.get(param).getAndSet(0), blockCountMap.get(param).getAndSet(0)
171.                     ));
172.                 }
173.                 System.out.println("=============================");
174.                 if (seconds-- <= 0) {
175.                     stop = true;
176.                 }
177.             }
178.             long cost = System.currentTimeMillis() - start;
179.             System.out.println("Time cost: " + cost + " ms");
180.             System.exit(0);
181.         }
182.     }
184.     final class RunTask implements Runnable {
185.         @Override
186.         public void run() {
187.             while (!stop) {
188.                 Entry entry = null;
189.                 T param = generateParam();
190.                 try {
191.                     entry = SphU.entry(resourceName, EntryType.IN, 1, param);
192.                     //Add pass for parameter.
193.                     passFor(param);
194.                 } catch (BlockException e) {
195.                     //block.incrementAndGet();
196.                     blockFor(param);
197.                 } catch (Exception ex) {
198.                     //biz exception
199.                     ex.printStackTrace();
200.                 } finally {
201.                     //total.incrementAndGet();
202.                     if (entry != null) {
203.                         entry.exit(1, param);
204.                     }
205.                 }
206.                 sleep(ThreadLocalRandom.current().nextInt(0, 10));
207.             }
208.         }
209.     }
211.     //Pick one of provided parameters randomly.
212.     private T generateParam() {
213.         int i = ThreadLocalRandom.current().nextInt(0, params.length);
214.         return params[i];
215.     }
217.     private void passFor(T param) {
218.         passCountMap.get(param).incrementAndGet();
219.     }
221.     private void blockFor(T param) {
222.         blockCountMap.get(param).incrementAndGet();
223.     }
225.     private void sleep(int timeMs) {
226.         try {
227.             TimeUnit.MILLISECONDS.sleep(timeMs);
228.         } catch (InterruptedException e) {
229.         }
230.     }
231. }

复制代码

四.参数限流是如何进行数据统计
由于参数限流的数据统计需要细化到参数值的维度，所以使用参数限流时需要注意OOM问题。比如根据用户ID进行限流，且用户数量有几千万，那么CacheMap将会包含几千万个不会被移除的键值对，而且会随着进程运行时间的增长而不断增加，最后可能会导致OOM。

1. @Spi(order = -3000)
2. public class ParamFlowSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
3.     @Override
4.     public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count,
5.         boolean prioritized, Object... args) throws Throwable {
6.         //1.如果没配置参数限流规则，直接触发下一个Slot
7.         if (!ParamFlowRuleManager.hasRules(resourceWrapper.getName())) {
8.             fireEntry(context, resourceWrapper, node, count, prioritized, args);
9.             return;
10.         }
11.         //2.如果配置了参数限流规则，则调用ParamFlowSlot的checkFlow()方法，该方法执行完成后再触发下一个Slot
12.         checkFlow(resourceWrapper, count, args);
13.         fireEntry(context, resourceWrapper, node, count, prioritized, args);
14.     }
16.     @Override
17.     public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
18.         fireExit(context, resourceWrapper, count, args);
19.     }
21.     void applyRealParamIdx(/*@NonNull*/ ParamFlowRule rule, int length) {
22.         int paramIdx = rule.getParamIdx();
23.         if (paramIdx < 0) {
24.             if (-paramIdx <= length) {
25.                 rule.setParamIdx(length + paramIdx);
26.             } else {
27.                 //Illegal index, give it a illegal positive value, latter rule checking will pass.
28.                 rule.setParamIdx(-paramIdx);
29.             }
30.         }
31.     }
33.     void checkFlow(ResourceWrapper resourceWrapper, int count, Object... args) throws BlockException {
34.         //1.如果没传递参数，则直接放行，代表不做参数限流逻辑
35.         if (args == null) {
36.             return;
37.         }
38.         //2.如果没给resourceWrapper这个资源配置参数限流规则，则直接放行
39.         if (!ParamFlowRuleManager.hasRules(resourceWrapper.getName())) {
40.             return;
41.         }
42.         //3.获取此资源的全部参数限流规则，规则可能会有很多个，所以是个List
43.         List<ParamFlowRule> rules = ParamFlowRuleManager.getRulesOfResource(resourceWrapper.getName());
44.         //4.遍历获取到的参数限流规则
45.         for (ParamFlowRule rule : rules) {
46.             //进行参数验证
47.             applyRealParamIdx(rule, args.length);
48.             //Initialize the parameter metrics.
49.             ParameterMetricStorage.initParamMetricsFor(resourceWrapper, rule);
50.             //进行验证的核心方法：检查当前规则是否允许通过此请求，如果不允许，则抛出ParamFlowException异常
51.             if (!ParamFlowChecker.passCheck(resourceWrapper, rule, count, args)) {
52.                 String triggeredParam = "";
53.                 if (args.length > rule.getParamIdx()) {
54.                     Object value = args[rule.getParamIdx()];
55.                     //Assign actual value with the result of paramFlowKey method
56.                     if (value instanceof ParamFlowArgument) {
57.                         value = ((ParamFlowArgument) value).paramFlowKey();
58.                     }
59.                     triggeredParam = String.valueOf(value);
60.                 }
61.                 throw new ParamFlowException(resourceWrapper.getName(), triggeredParam, rule);
62.             }
63.         }
64.     }
65. }

复制代码

五.参数限流验证请求的流程图总结

 
2.@SentinelResource注解的使用和实现
(1)@SentinelResource注解的使用
(2)@SentinelResource注解和实现
 
(1)@SentinelResource注解的使用
一.引入Sentinel Spring Boot Starter依赖

1. //Rule checker for parameter flow control.
2. public final class ParamFlowChecker {
3.     public static boolean passCheck(ResourceWrapper resourceWrapper, /*@Valid*/ ParamFlowRule rule, /*@Valid*/ int count, Object... args) {
4.         if (args == null) {
5.             return true;
6.         }
7.         //1.判断参数索引是否合法，这个就是配置参数限流时设置的下标，从0开始，也就是对应args里的下标
8.         //比如0就代表args数组里的第一个参数，如果参数不合法直接放行，相当于参数限流没生效  
9.         int paramIdx = rule.getParamIdx();
10.         if (args.length <= paramIdx) {
11.             return true;
12.         }
13.         //2.判断参数值是不是空，如果是空直接放行
14.         //Get parameter value.
15.         Object value = args[paramIdx];
16.         //Assign value with the result of paramFlowKey method
17.         if (value instanceof ParamFlowArgument) {
18.             value = ((ParamFlowArgument) value).paramFlowKey();
19.         }
20.         //If value is null, then pass
21.         if (value == null) {
22.             return true;
23.         }
24.         //3.集群限流
25.         if (rule.isClusterMode() && rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {
26.             return passClusterCheck(resourceWrapper, rule, count, value);
27.         }
28.         //4.单机限流
29.         return passLocalCheck(resourceWrapper, rule, count, value);
30.     }
32.     private static boolean passLocalCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int count, Object value) {
33.         try {
34.             if (Collection.class.isAssignableFrom(value.getClass())) {//基本类型
35.                 for (Object param : ((Collection)value)) {
36.                     if (!passSingleValueCheck(resourceWrapper, rule, count, param)) {
37.                         return false;
38.                     }
39.                 }
40.             } else if (value.getClass().isArray()) {//数组类型
41.                 int length = Array.getLength(value);
42.                 for (int i = 0; i < length; i++) {
43.                     Object param = Array.get(value, i);
44.                     if (!passSingleValueCheck(resourceWrapper, rule, count, param)) {
45.                         return false;
46.                     }
47.                 }
48.             } else {//其他类型，也就是引用类型
49.                 return passSingleValueCheck(resourceWrapper, rule, count, value);
50.             }
51.         } catch (Throwable e) {
52.             RecordLog.warn("[ParamFlowChecker] Unexpected error", e);
53.         }
54.         return true;
55.     }
57.     static boolean passSingleValueCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int acquireCount, Object value) {
58.         if (rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {//类型是QPS
59.             if (rule.getControlBehavior() == RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER) {
60.                 //流控效果为排队等待
61.                 return passThrottleLocalCheck(resourceWrapper, rule, acquireCount, value);
62.             } else {
63.                 //流控效果为直接拒绝
64.                 return passDefaultLocalCheck(resourceWrapper, rule, acquireCount, value);
65.             }
66.         } else if (rule.getGrade() == RuleConstant.FLOW_GRADE_THREAD) {//类型是Thread
67.             Set<Object> exclusionItems = rule.getParsedHotItems().keySet();
68.             long threadCount = getParameterMetric(resourceWrapper).getThreadCount(rule.getParamIdx(), value);
69.             if (exclusionItems.contains(value)) {
70.                 int itemThreshold = rule.getParsedHotItems().get(value);
71.                 return ++threadCount <= itemThreshold;
72.             }
73.             long threshold = (long)rule.getCount();
74.             return ++threadCount <= threshold;
75.         }
76.         return true;
77.     }
78.     ...
79. }

复制代码

二.为方法添加@SentinelResource注解
下面的代码为sayHello()方法添加了@SentinelResource注解，并指定了资源名称为sayHello以及熔断降级时的回调方法fallback()。这样在请求sayHello()方法后，就可以在Sentinel Dashboard上看到此资源，然后就可以针对此资源进行一系列的规则配置了。

1. //Rule checker for parameter flow control.
2. public final class ParamFlowChecker {
3.     ...
4.     static boolean passThrottleLocalCheck(ResourceWrapper resourceWrapper, ParamFlowRule rule, int acquireCount, Object value) {
5.         ParameterMetric metric = getParameterMetric(resourceWrapper);
6.         CacheMap<Object, AtomicLong> timeRecorderMap = metric == null ? null : metric.getRuleTimeCounter(rule);
7.         if (timeRecorderMap == null) {
8.             return true;
9.         }
10.         //Calculate max token count (threshold)
11.         Set<Object> exclusionItems = rule.getParsedHotItems().keySet();
12.         long tokenCount = (long)rule.getCount();
13.         if (exclusionItems.contains(value)) {
14.             tokenCount = rule.getParsedHotItems().get(value);
15.         }
16.         if (tokenCount == 0) {
17.             return false;
18.         }
19.         long costTime = Math.round(1.0 * 1000 * acquireCount * rule.getDurationInSec() / tokenCount);
20.         while (true) {
21.             long currentTime = TimeUtil.currentTimeMillis();
22.             AtomicLong timeRecorder = timeRecorderMap.putIfAbsent(value, new AtomicLong(currentTime));
23.             if (timeRecorder == null) {
24.                 return true;
25.             }
26.             //AtomicLong timeRecorder = timeRecorderMap.get(value);
27.             long lastPassTime = timeRecorder.get();
28.             long expectedTime = lastPassTime + costTime;
29.             if (expectedTime <= currentTime || expectedTime - currentTime < rule.getMaxQueueingTimeMs()) {
30.                 AtomicLong lastPastTimeRef = timeRecorderMap.get(value);
31.                 if (lastPastTimeRef.compareAndSet(lastPassTime, currentTime)) {
32.                     long waitTime = expectedTime - currentTime;
33.                     if (waitTime > 0) {
34.                         lastPastTimeRef.set(expectedTime);
35.                         try {
36.                             TimeUnit.MILLISECONDS.sleep(waitTime);
37.                         } catch (InterruptedException e) {
38.                             RecordLog.warn("passThrottleLocalCheck: wait interrupted", e);
39.                         }
40.                     }
41.                     return true;
42.                 } else {
43.                     Thread.yield();
44.                 }
45.             } else {
46.                 return false;
47.             }
48.         }
49.     }
51.     private static ParameterMetric getParameterMetric(ResourceWrapper resourceWrapper) {
52.         //Should not be null.
53.         return ParameterMetricStorage.getParamMetric(resourceWrapper);
54.     }
55. }

复制代码

(2)@SentinelResource注解和实现
利用Spring AOP拦截@SentinelResource注解，最后调用SphU.entry()方法来进行处理。
[code]//Aspect for methods with {@link SentinelResource} annotation.@Aspectpublic class SentinelResourceAspect extends AbstractSentinelAspectSupport {    //SentinelResource注解    @Pointcut("@annotation(com.alibaba.csp.sentinel.annotation.SentinelResource)")    public void sentinelResourceAnnotationPointcut() {    }    @Around("sentinelResourceAnnotationPointcut()")    public Object invokeResourceWithSentinel(ProceedingJoinPoint pjp) throws Throwable {        //获取方法        Method originMethod = resolveMethod(pjp);        //获取方法上的SentinelResource注解，有了这个注解，就可以获取到注解的各种属性值了        SentinelResource annotation = originMethod.getAnnotation(SentinelResource.class);        if (annotation == null) {            //Should not go through here.            throw new IllegalStateException("Wrong state for SentinelResource annotation");        }        //获取资源名称        String resourceName = getResourceName(annotation.value(), originMethod);        //获取资源类型        EntryType entryType = annotation.entryType();        int resourceType = annotation.resourceType();        //创建一个Entry对象，通过SphU.entry(resourceName)将当前方法纳入Sentinel的保护体系        //如果当前资源的调用未触发任何Sentinel规则，则正常执行被拦截的方法，否则将执行对应的限流、熔断降级等处理逻辑        Entry entry = null;        try {            entry = SphU.entry(resourceName, resourceType, entryType, pjp.getArgs());            return pjp.proceed();        } catch (BlockException ex) {            //发生异常时，通过反射执行在注解中设置的降级方法            return handleBlockException(pjp, annotation, ex);        } catch (Throwable ex) {            Class