关于两种限流模式

Posted 2020-08-07 穿林度水

流量预警和限流方案中，比较常用的有两种。第一种滑窗模式，通过统计一段时间内的访问次数来进行控制，访问次数达到的某个峰值时进行限流。第二种为并发用户数模式，通过控制最大并发用户数，来达到流量控制的目的。下面来简单分析下两种的优缺点。

 

1、滑窗模式

模式分析:

在每次有访问进来时，我们判断前N个单位时间内的总访问量是否超过了设置的阈值，并对当前时间片上的请求数+1。

上图每一个格式表示一个固定的时间（比如1s），每个格子一个计数器，我们要获取前5s的请求量，就是对当前时间片i ~ i-4的时间片上计数器进行累加。

这种模式的实现的方式更加契合流控的本质意义。理解较为简单。但由于访问量的不可预见性，会发生单位时间的前半段大量请求涌入，而后半段则拒绝所有请求的情况。（通常，需要可以将单位时间切的足够的小来缓解 ）其次，我们很难确定这个阈值设置在多少比较合适，只能通过经验或者模拟（如压测）来进行估计，即使是压测也很难估计的准确。集群部署中每台机器的硬件参 数不同，可能导致我们需要对每台机器的阈值设置的都不尽相同。同一台机子在不同的时间点的系统压力也不一样（比如晚上还有一些任务，或其他的一些业务操作 的影响），能够承受的最大阈值也不尽相同，我们无法考虑的周全。

所以滑窗模式通常适用于对某一资源的保护的需求上（或者说是承诺比较合适：我对某一接口的提供者承诺过，最高调用量不超过XX），如对db的保护，对某一服务的调用的控制上。

代码实现思路：

每一个时间片（单位时间）就是一个独立的计数器，用以数组保存。将当前时间以某种方式（比如取模）映射到数组的一项中。每次访问先对当前时间片上的计数器+1，再计算前N个时间片的访问量总合，超过阈值则限流。

 

 

Java代码  

1. /** 
2.  * 滑窗的实现 
3.  * @author shimig 
4.  * 
5.  */  
6. public class SlidingWindow {  
7.   
8.     /* 循环队列 */  
9.     private volatile AtomicInteger[] timeSlices;  
10.     /* 队列的总长度  */  
11.     private volatile int timeSliceSize;  
12.     /* 每个时间片的时长 */  
13.     private volatile int timeMillisPerSlice;  
14.     /* 窗口长度 */  
15.     private volatile int windowSize;  
16.       
17.     /* 当前所使用的时间片位置 */  
18.     private AtomicInteger cursor = new AtomicInteger(0);  
19.       
20.     public SlidingWindow(int timeMillisPerSlice, int windowSize) {  
21.         this.timeMillisPerSlice = timeMillisPerSlice;  
22.         this.windowSize = windowSize;  
23.         // 保证存储在至少两个window  
24.         this.timeSliceSize = windowSize * 2 + 1;  
25.     }  
26.       
27.     /** 
28.      * 初始化队列，由于此初始化会申请一些内容空间，为了节省空间，延迟初始化 
29.      */  
30.     private void initTimeSlices() {  
31.         if (timeSlices != null) {  
32.             return;  
33.         }  
34.         // 在多线程的情况下，会出现多次初始化的情况，没关系  
35.         // 我们只需要保证，获取到的值一定是一个稳定的，所有这里使用先初始化，最后赋值的方法  
36.         AtomicInteger[] localTimeSlices = new AtomicInteger[timeSliceSize];  
37.         for (int i = 0; i < timeSliceSize; i++) {  
38.             localTimeSlices[i] = new AtomicInteger(0);  
39.         }  
40.         timeSlices = localTimeSlices;  
41.     }  
42.       
43.     private int locationIndex() {  
44.         long time = System.currentTimeMillis();  
45.         return (int) ((time / timeMillisPerSlice) % timeSliceSize);  
46.     }  
47.       
48.     /** 
49.      * <p>对时间片计数+1，并返回窗口中所有的计数总和 
50.      * <p>该方法只要调用就一定会对某个时间片进行+1 
51.      *  
52.      * @return 
53.      */  
54.     public int incrementAndSum() {  
55.         initTimeSlices();  
56.         int index = locationIndex();  
57.         int sum = 0;  
58.         // cursor等于index，返回true  
59.         // cursor不等于index，返回false，并会将cursor设置为index  
60.         int oldCursor = cursor.getAndSet(index);  
61.         if (oldCursor == index) {  
62.             // 在当前时间片里继续+1  
63.             sum += timeSlices[index].incrementAndGet();  
64.         } else {  
65.             // 可能有其他thread已经置过1，问题不大  
66.             timeSlices[index].set(1);  
67.             // 清零，访问量不大时会有时间片跳跃的情况  
68.             clearBetween(oldCursor, index);  
69.             // sum += 0;  
70.         }  
71.         for (int i = 1; i < windowSize; i++) {  
72.             sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get();  
73.         }  
74.         return sum;  
75.     }  
76.       
77.     /** 
78.      * 判断是否允许进行访问，未超过阈值的话才会对某个时间片+1 
79.      *  
80.      * @param threshold 
81.      * @return 
82.      */  
83.     public boolean allow(int threshold) {  
84.         initTimeSlices();  
85.         int index = locationIndex();  
86.         int sum = 0;  
87.         // cursor不等于index，将cursor设置为index  
88.         int oldCursor = cursor.getAndSet(index);  
89.         if (oldCursor != index) {  
90.             // 可能有其他thread已经置过1，问题不大  
91.             timeSlices[index].set(0);  
92.             // 清零，访问量不大时会有时间片跳跃的情况  
93.             clearBetween(oldCursor, index);  
94.         }  
95.         for (int i = 1; i < windowSize; i++) {  
96.             sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get();  
97.         }  
98.           
99.         // 阈值判断  
100.         if (sum <= threshold) {  
101.             // 未超过阈值才+1  
102.             sum += timeSlices[index].incrementAndGet();  
103.             return true;  
104.         }  
105.         return false;  
106.     }  
107.       
108.     /** 
109.      * <p>将fromIndex~toIndex之间的时间片计数都清零 
110.      * <p>极端情况下，当循环队列已经走了超过1个timeSliceSize以上，这里的清零并不能如期望的进行 
111.      *  
112.      * @param fromIndex 不包含 
113.      * @param toIndex 不包含 
114.      */  
115.     private void clearBetween(int fromIndex, int toIndex) {  
116.         for (int index = (fromIndex + 1) % timeSliceSize; index != toIndex; index = (index + 1) % timeSliceSize) {  
117.             timeSlices[index].set(0);  
118.         }  
119.     }  
120.       
121. }  

 

2、并发用户数模式

模式分析：

每次操作执行时，我们通过判断当前正在执行的访问数是否超过某个阈值在决定是否限流。

 

该模式看着思路比较的另类，但却有其独到之处。实际上我们限流的根本是为了保护资源，防止系统接受的请求过多，应接不暇，拖慢系统中其他接口的服务，造成雪崩。我们真正需要关心的是那些运行中的请求，而那些已经完成的请求已是过去时，不再是需要关心的了。

我们来看看其阈值的计算方式，对于一个请求来说，响应时间rt、qps是一个比较容易获取的参数，那么我们这样计算：qps/1000*rt。

此外，一个应用往往是个复杂的系统，提供的服务或者暴露的请求、资源不止一个。内部GC、定时任务的执行、其他服务访问的骤增，外部依赖方、db的抖动，抑或是代码中不经意间的一个bug。都可能导致响应时间的变化，导致系统性能容量的改变 。而这种模式，则能恰如其分的自动做出调整，当系统不适时，rt增加，会自动的对qps做出适应。

代码实现思路:

当访问开始时，我们对当前计数器（原子计数器）+1，当完成时，-1。该计数器即为当前正在执行的请求数。只需判断这个计数器是否超过阈值即可。