限流算法的理解和应用场景和实现[临界点处理]

淡淡的烟草味﹌ 2022-05-30 02:43 112阅读 0赞

在开发高并发系统时，有三把利器来保护系统：缓存、降级和限流。一下有几种限流的方法可以参考。

### 信号量和令牌桶的区别: ###

信号量限制的是并发,资源. 令牌桶如果耗时比较高的话,并发可能会比较大. 限制的是 qps.

### 计数器法 ###

计数器法是限流算法里最简单也是最容易实现的一种算法。比如我们规定，对于A接口来说，我们1分钟的访问次数不能超过100个。那么我们可以这么做：在一开 始的时候，我们可以设置一个计数器counter，每当一个请求过来的时候，counter就加1，如果counter的值大于100并且该请求与第一个 请求的间隔时间还在1分钟之内，那么说明请求数过多；如果该请求与第一个请求的间隔时间大于1分钟，且counter的值还在限流范围内，那么就重置 counter，具体算法的示意图如下：

[![2016-09-01\_20:31:28.jpg][2016-09-01_20_31_28.jpg]][2016-09-01_20_31_28.jpg]

#  #

这个算法虽然简单，但是有一个十分致命的问题，那就是临界问题，我们看下图：

[![2016-09-01\_20:35:21.jpg][2016-09-01_20_35_21.jpg]][2016-09-01_20_35_21.jpg]

从上图中我们可以看到，假设有一个恶意用户，他在0:59时，瞬间发送了100个请求，并且1:00又瞬间发送了100个请求，那么其实这个用户在 1秒里面，瞬间发送了200个请求。我们刚才规定的是1分钟最多100个请求，也就是每秒钟最多1.7个请求，用户通过在时间窗口的重置节点处突发请求， 可以瞬间超过我们的速率限制。用户有可能通过算法的这个漏洞，瞬间压垮我们的应用。

聪明的朋友可能已经看出来了，刚才的问题其实是因为我们统计的精度太低。那么如何很好地处理这个问题呢？或者说，如何将临界问题的影响降低呢？我们可以看下面的滑动窗口算法。

public class CounterDemo {  
        public long timeStamp = getNowTime();  
        public int reqCount = 0;  
        public final int limit = 100; // 时间窗口内最大请求数  
        public final long interval = 1000; // 时间窗口ms  
        public boolean grant() {  
            long now = getNowTime();  
            if (now < timeStamp + interval) {  
                // 在时间窗口内  
                reqCount++;  
                // 判断当前时间窗口内是否超过最大请求控制数  
                return reqCount <= limit;  
            }  
            else {  
                timeStamp = now;  
                // 超时后重置  
                reqCount = 1;  
                return true;  
            }  
        }  
    }

# 滑动窗口 #

滑动窗口，又称rolling window。为了解决这个问题，我们引入了滑动窗口算法。如果学过TCP网络协议的话，那么一定对滑动窗口这个名词不会陌生。下面这张图，很好地解释了滑动窗口算法：

[![2016-09-01\_20:42:46.jpg][2016-09-01_20_42_46.jpg]][2016-09-01_20_42_46.jpg]

在上图中，整个红色的矩形框表示一个时间窗口，在我们的例子中，一个时间窗口就是一分钟。然后我们将时间窗口进行划分，比如图中，我们就将滑动窗口 划成了6格，所以每格代表的是10秒钟。每过10秒钟，我们的时间窗口就会往右滑动一格。每一个格子都有自己独立的计数器counter，比如当一个请求 在0:35秒的时候到达，那么0:30~0:39对应的counter就会加1。

那么滑动窗口怎么解决刚才的临界问题的呢？我们可以看上图，0:59到达的100个请求会落在灰色的格子中，而1:00到达的请求会落在橘黄色的格 子中。当时间到达1:00时，我们的窗口会往右移动一格，那么此时时间窗口内的总请求数量一共是200个，超过了限定的100个，所以此时能够检测出来触 发了限流。

我再来回顾一下刚才的计数器算法，我们可以发现，计数器算法其实就是滑动窗口算法。只是它没有对时间窗口做进一步地划分，所以只有1格。

由此可见，当滑动窗口的格子划分的越多，那么滑动窗口的滚动就越平滑，限流的统计就会越精确。

public class LeakyDemo {  
        public long timeStamp = getNowTime();  
        public int capacity; // 桶的容量  
        public int rate; // 水漏出的速度  
        public int water; // 当前水量(当前累积请求数)  
        public boolean grant() {  
            long now = getNowTime();  
            water = max(0, water - (now - timeStamp) * rate); // 先执行漏水，计算剩余水量  
            timeStamp = now;  
            if ((water + 1) < capacity) {  
                // 尝试加水,并且水还未满  
                water += 1;  
                return true;  
            }  
            else {  
                // 水满，拒绝加水  
                return false;  
            }  
        }  
    }

# 先看看漏桶算法(Leaky bucket) #

![SouthEast][]

![SouthEast 1][]

如图所示,很明显从原来两个流量(12mbps 和2mbps)限流成了 3mbps.

## 实现:    ##

![SouthEast 2][]

一个比较简单实现是: n 个线程这种先把数据流量放置到一个队列里(或者 一个接口拆成1个队列,分而治之), 然后另外一个线程从里面获取数据,请求.

## 应用场景:     ##

异步化的调用比较好, 同步化的调用的话,就需要搞成类似 reactor 模式的形式,每个数据包需要有 seq\_no 的概念(tcp,dubbo 异步传输).

# 再看看令牌桶(Token bucket): #

Guava官方文档-RateLimiter类![Center][]

public class TokenBucketDemo {  
        public long timeStamp = getNowTime();  
        public int capacity; // 桶的容量  
        public int rate; // 令牌放入速度  
        public int tokens; // 当前令牌数量  
        public boolean grant() {  
            long now = getNowTime();  
            // 先添加令牌  
            tokens = min(capacity, tokens + (now - timeStamp) * rate);   
            timeStamp = now;  
            if (tokens < 1) {  
                // 若不到1个令牌,则拒绝  
                return false;  
            }  
            else {  
                // 还有令牌，领取令牌  
                tokens -= 1;  
                return true;  
            }  
        }  
    }

# 使用Guava的RateLimiter进行限流控制 #

Guava是google提供的java扩展类库，其中的限流工具类RateLimiter采用的就是令牌桶算法。RateLimiter 从概念上来讲，速率限制器会在可配置的速率下分配许可证，如果必要的话，每个acquire() 会阻塞当前线程直到许可证可用后获取该许可证，一旦获取到许可证，不需要再释放许可证。通俗的讲RateLimiter会按照一定的频率往桶里扔令牌，线程拿到令牌才能执行，比如你希望自己的应用程序QPS不要超过1000，那么RateLimiter设置1000的速率后，就会每秒往桶里扔1000个令牌。例如我们需要处理一个任务列表，但我们不希望每秒的任务提交超过两个，此时可以采用如下方式：

![Center 1][]  
有一点很重要，那就是请求的许可数从来不会影响到请求本身的限制（调用acquire(1) 和调用acquire(1000) 将得到相同的限制效果，如果存在这样的调用的话），但会影响下一次请求的限制，也就是说，如果一个高开销的任务抵达一个空闲的RateLimiter，它会被马上许可，但是下一个请求会经历额外的限制，从而来偿付高开销任务。注意：RateLimiter 并不提供公平性的保证。

\[java\]  [view plain][]  [copy][view plain]

1.  public class <span style="font-size:14px;">RateLimiter</span>\{  
2.    public double acquire() \{  
3.          return acquire(1);  
4.      \}  
5.    
6.   public double acquire(int permits) \{  
7.          checkPermits(permits);  //检查参数是否合法（是否大于0）  
8.          long microsToWait;  
9.          synchronized (mutex) \{ //应对并发情况需要同步  
10.             microsToWait = reserveNextTicket(permits, readSafeMicros()); //获得需要等待的时间   
11.         \}  
12.         ticker.sleepMicrosUninterruptibly(microsToWait); //等待，当未达到限制时，microsToWait为0  
13.         return 1.0 \* microsToWait / TimeUnit.SECONDS.toMicros(1L);  
14.     \}  
15.   
16. private long reserveNextTicket(double requiredPermits, long nowMicros) \{  
17.         resync(nowMicros); //补充令牌  
18.         long microsToNextFreeTicket = nextFreeTicketMicros - nowMicros;  
19.         double storedPermitsToSpend = Math.min(requiredPermits, this.storedPermits); //获取这次请求消耗的令牌数目  
20.         double freshPermits = requiredPermits - storedPermitsToSpend;  
21.   
22.         long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)  
23.                 + (long) (freshPermits \* stableIntervalMicros);   
24.   
25.         this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;  
26.         this.storedPermits -= storedPermitsToSpend; // 减去消耗的令牌  
27.         return microsToNextFreeTicket;  
28.     \}  
29.   
30. private void resync(long nowMicros) \{  
31.         // if nextFreeTicket is in the past, resync to now  
32.         if (nowMicros > nextFreeTicketMicros) \{  
33.             storedPermits = Math.min(maxPermits,  
34.                     storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);  
35.             nextFreeTicketMicros = nowMicros;  
36.         \}  
37.     \}  
38. \}

# 四、使用Semphore进行并发流控 #

Java 并发库的Semaphore 可以很轻松完成信号量控制，Semaphore可以控制某个资源可被同时访问的个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。单个信号量的Semaphore对象可以实现互斥锁的功能，并且可以是由一个线程获得了“锁”，再由另一个线程释放“锁”，这可应用于死锁恢复的一些场合。下面的Demo中申明了一个只有5个许可的Semaphore，而有20个线程要访问这个资源，通过acquire()和release()获取和释放访问许可：

![Center 2][]

![Center 3][]

最后：进行限流控制还可以有很多种方法，针对不同的场景各有优劣，例如通过AtomicLong计数器控制、使用MQ消息队列进行流量消峰等等。

[2016-09-01_20_31_28.jpg]: http://7xrd8h.com1.z0.glb.clouddn.com/2016-09-01_20:31:28.jpg
[2016-09-01_20_35_21.jpg]: http://7xrd8h.com1.z0.glb.clouddn.com/2016-09-01_20:35:21.jpg
[2016-09-01_20_42_46.jpg]: http://7xrd8h.com1.z0.glb.clouddn.com/2016-09-01_20:42:46.jpg
[SouthEast]: /images/20220530/6e8d7da863f3487894e1c96501bb5a9f.png
[SouthEast 1]: /images/20220530/8725425be6744aa0b51c3a56905f5c48.png
[SouthEast 2]: /images/20220530/df86b6d5f8d34cceafa5327463affe54.png
[Center]: /images/20220530/15030394d3a64cfa8f65220682484fbc.png
[Center 1]: /images/20220530/89f915599fc04ee7a2ef1f8eef9df92e.png
[view plain]: http://blog.csdn.net/fei33423/article/details/79294496#
[Center 2]: /images/20220530/995f94d926c348c7872e77db4a026987.png
[Center 3]: /images/20220530/2cb5457a2ebd4fdb922ee19d3a4b5d60.png