精讲高并发核心编程，限流原理与实战，限流策略原理与参考实现

愿天堂没有BUG

发布于 2022-10-28 11:47:13

2720

发布于 2022-10-28 11:47:13

文章被收录于专栏：愿天堂没有BUG（公众号同名）

限流原理与实战

在通信领域中，限流技术（Time Limiting）被用来控制网络接口收发通信数据的速率，实现通信时的优化性能、较少延迟和提高带宽等。

互联网领域中借鉴了通信领域的限流概念，用来控制在高并发、大流量的场景中对服务接口请求的速率，比如双十一秒杀、抢购、抢票、抢单等场景。

举一个具体的例子，假设某个接口能够扛住的QPS为10 000，这时有20 000个请求进来，经过限流模块，会先放10 000个请求，其余的请求会阻塞一段时间。不简单粗暴地返回404，让客户端重试，同时又能起到流量削峰的作用。

每个API接口都是有访问上限的，当访问频率或者并发量超过其承受范围时，就必须通过限流来保证接口的可用性或者降级可用性，给接口安装上保险丝，以防止非预期的请求对系统压力过大而引起系统瘫痪。

接口限流的算法主要有3种，分别是计数器限流、漏桶算法和令牌桶算法。接下来为大家一一介绍。

限流策略原理与参考实现

在高并发访问的情况下，通常会通过限流的手段来控制流量问题，以保证服务器处于正常压力下。

首先给大家介绍3种常见的限流策略。

3种限流策略：计数器、漏桶和令牌桶

限流的手段通常有计数器、漏桶和令牌桶。注意限流和限速（所有请求都会处理）的差别，视业务场景而定。

（1）计数器：在一段时间间隔内（时间窗），处理请求的最大数量固定，超过部分不做处理。

（2）漏桶：漏桶大小固定，处理速度固定，但请求进入的速度不固定（在突发情况请求过多时，会丢弃过多的请求）。

（3）令牌桶：令牌桶的大小固定，令牌的产生速度固定，但是小耗令牌（请求）的速度不固定（可以应对某些时间请求过多的情况）。每个请求都会从令牌桶中取出令牌，如果没有令牌，就丢弃这次请求。

计数器限流原理和Java参考实现

计数器限流的原理非常简单：在一个时间窗口（间隔）内，所处理的请求的最大数量是有限制的，对超过限制的部分请求不做处理。

下面的代码是计数器限流算法的一个简单的演示实现和测试用例。

package com.crazymaker.springcloud.ratelimit;...//计速器，限速@Slf4jpublic class CounterLimiter{ //起始时间 private static long startTime = System.currentTimeMillis(); //时间区间的时间间隔毫秒 private static long interval = ; //每秒限制数量 private static long maxCount = ; //累加器 private static AtomicLong accumulator = new AtomicLong(); //计数判断，是否超出限制 private static long tryAcquire(long taskId, int turn) { long nowTime = System.currentTimeMillis(); //在时间区间之内 if (nowTime < startTime + interval) { long count = accumulator.incrementAndGet(); if (count <= maxCount) { return count; } else { return -count; } } else { //在时间区间之外 synchronized (CounterLimiter.class) { log.info("新时间区到了,taskId{}, turn {}..", taskId, turn); //再一次判断，防止重复初始化 if (nowTime > startTime + interval) { accumulator.set(); startTime = nowTime; } } return ; } } //线程池，用于多线程模拟测试 private ExecutorService pool = Executors.newFixedThreadPool(); @Test public void testLimit() { //被限制的次数 AtomicInteger limited = new AtomicInteger(); //线程数 final int threads = ; //每条线程的执行轮数 final int turns = ; //同步器 CountDownLatch countDownLatch = new CountDownLatch(threads); long start = System.currentTimeMillis(); for (int i = ; i < threads; i++) { pool.submit(() -> { try { for (int j = ; j < turns; j++) { long taskId = Thread.currentThread().getId(); long index = tryAcquire(taskId, j); if (index <= ) { //被限制的次数累积 limited.getAndIncrement(); } Thread.sleep(); } } catch (Exception e) { e.printStackTrace(); } //等待所有线程结束 countDownLatch.countDown(); }); } try { countDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } float time = (System.currentTimeMillis() - start) / F; //输出统计结果 log.info("限制的次数为：" + limited.get() + ",通过的次数为：" + (threads *turns - limited.get())); log.info("限制的比例为：" + (float) limited.get() / (float) (threads *turns)); log.info("运行的时长为：" + time); }}

以上代码使用两条线程，每条线程各运行20次，每一次运行休眠200毫秒，总计耗时4秒，运行40次，限流的输出结果具体如下：

[pool-2-thread-2] INFO c.c.s.ratelimit.CounterLimiter - 新时间区到了,taskId16, turn 5..[pool-2-thread-1] INFO c.c.s.ratelimit.CounterLimiter - 新时间区到了,taskId15, turn 5..[pool-2-thread-2] INFO c.c.s.ratelimit.CounterLimiter - 新时间区到了,taskId16, turn 10..[pool-2-thread-2] INFO c.c.s.ratelimit.CounterLimiter - 新时间区到了,taskId16, turn 15..[main] INFO c.c.s.ratelimit.CounterLimiter - 限制的次数为：32,通过的次数为：8[main] INFO c.c.s.ratelimit.CounterLimiter - 限制的比例为：0.8[main] INFO c.c.s.ratelimit.CounterLimiter - 运行的时长为：4.104

大家可以自行调整参数，运行以上自验证程序并观察实验结果，体验一下计数器限流的效果。

漏桶限流原理和Java参考实现

漏桶限流的基本原理：水（对应请求）从进水口进入漏桶里，漏桶以一定的速度出水（请求放行），当水流入的速度过大时，桶内的总水量大于桶容量会直接溢出，请求被拒绝，如图9-1所示。

大致的漏桶限流规则如下：

（1）水通过进水口（对应客户端请求）以任意速率流入漏桶。

（2）漏桶的容量是固定的，出水（放行）速率也是固定的。

（3）漏桶容量是不变的，如果处理速度太慢，桶内水量会超出桶的容量，后面流入的水就会溢出，表示请求拒绝。

图9-1 漏桶原理示意图

漏桶的Java参考实现代码如下：

package com.crazymaker.springcloud.ratelimit;//省略import//漏桶限流@Slf4jpublic class LeakBucketLimiter{ //计算的起始时间 private static long lastOutTime = System.currentTimeMillis(); //流出速率每秒2次 private static long rate = 2; //剩余的水量 private static long water = 0; //返回值说明 //false：没有被限制到 //true：被限流 public static synchronized boolean tryAcquire(long taskId, int turn) { long nowTime = System.currentTimeMillis(); //过去的时间 long pastTime = nowTime - lastOutTime; //漏出水量，按照恒定的速度不断流出 //漏出的水 = 过去的时间 *预设速率 long outWater = pastTime *rate / ; //剩余的水量 = 上次遗留的水量 - 漏出去的水 water = water - outWater; log.info("water {} pastTime {} outWater {} ",water, pastTime, outWater); //纠正剩余的水量 if (water < ) { water = ; } //若剩余的水量小于等于1，则放行 if (water <= 1) { //更新起始时间，为了下次使用 lastOutTime = nowTime; //增加遗留的水量 water ++; return false; } else { //剩余的水量太大，被限流 return true; } } //线程池，用于多线程模拟测试 private ExecutorService pool = Executors.newFixedThreadPool(10); //测试用例 @Test public void testLimit() { //测试用例太长，这里省略 //90%的测试用例代码与前面的计算器限流测试代码相同 //具体的源码可参见疯狂创客圈社群的开源库 }}

以下是使用两条线程，每条线程各运行20次，每一次运行休眠200毫秒，总计耗时4秒，运行40次，部分输出结果如下：

[pool-2-thread-1] INFO c.c.s.r.LeakBucketLimiter - water 0 pastTime 75 outWater 0...[pool-2-thread-1] INFO c.c.s.r.LeakBucketLimiter - water 1 pastTime 601 outWater 1...[pool-2-thread-1] INFO c.c.s.r.LeakBucketLimiter - water 2 pastTime 416 outWater 0[pool-2-thread-2] INFO c.c.s.r.LeakBucketLimiter - water 1 pastTime 601 outWater 1[pool-2-thread-1] INFO c.c.s.r.LeakBucketLimiter - water 2 pastTime 15 outWater 0[pool-2-thread-2] INFO c.c.s.r.LeakBucketLimiter - water 2 pastTime 201 outWater 0 [pool-2-thread-1] INFO c.c.s.r.LeakBucketLimiter - water 2 pastTime 216 outWater 0[main] INFO c.c.s.r.LeakBucketLimiter - 限制的次数为：32,通过的次数为：8[main] INFO c.c.s.r.LeakBucketLimiter - 限制的比例为：0.8[main] INFO c.c.s.r.LeakBucketLimiter - 运行的时长为：4.107

漏桶的出水速度固定，也就是请求放行速度是固定的。故漏桶不能有效应对突发流量，但是能起到平滑突发流量（整流）的作用。

令牌桶限流原理和Java参考实现

令牌桶算法以一个设定的速率产生令牌并放入令牌桶，每次用户请求都得申请令牌，如果令牌不足，就会拒绝请求。

在令牌桶算法中，新请求到来时会从桶里拿走一个令牌，如果桶内没有令牌可拿，就拒绝服务。当然，令牌的数量是有上限的。令牌的数量与时间和发放速率强相关，流逝的时间越长，会不断往桶里加入越多令牌，如果令牌发放的速度比申请速度快，令牌桶就会放满令牌，直到令牌占满整个令牌桶，如图9-2所示。

另外，令牌的发送速率可以设置，从而可以对突发流量进行有效的应对。

令牌桶限流大致的规则如下：

（1）进水口按照某个速度向桶中放入令牌。

（2）令牌的容量是固定的，但是放行的速度是不固定的，只要桶中还有剩余令牌，一旦请求过来就能申请成功，然后放行。

（3）如果令牌的发放速度慢于请求到来的速度，桶内就无令牌可领，请求就会被拒绝。

图9-2 令牌桶

令牌桶的Java参考实现代码如下：

package com.crazymaker.springcloud.ratelimit;...//令牌桶，限速@Slf4jpublic class TokenBucketLimiter{ //上一次令牌发放的时间 public long lastTime = System.currentTimeMillis(); //桶的容量 public int capacity = 2; //令牌生成速度个/秒 public int rate = ; //当前令牌的数量 public int tokens; //返回值说明 //false：没有被限制到 //true：被限流 public synchronized boolean tryAcquire(long taskId, int applyCount) { long now = System.currentTimeMillis();时间间隔单位为毫秒 //时间间隔，单位为毫秒 long gap = now - lastTime; //当前令牌数 tokens = Math.min(capacity, (int) (tokens + gap *rate/ )); log.info("tokens {} capacity {} gap {} ", tokens, capacity, gap); if (tokens < applyCount) { //若拿不到令牌，则拒绝 //log.info("被限流了.." + taskId + ", applyCount: " + applyCount); return true; } else { //还有令牌，领取令牌 tokens -= applyCount; lastTime = now; //log.info("剩余令牌.." + tokens); return false; } } //线程池，用于多线程模拟测试 private ExecutorService pool = Executors.newFixedThreadPool(10); @Test public void testLimit() { //被限制的次数 AtomicInteger limited = new AtomicInteger(0); //线程数 final int threads = 2; //每条线程的执行轮数 final int turns = 20; //同步器 CountDownLatch countDownLatch = new CountDownLatch(threads); long start = System.currentTimeMillis(); for (int i = 0; i < threads; i++) { pool.submit(() -> { try { for (int j = 0; j < turns; j++) { long taskId = Thread.currentThread().getId(); boolean intercepted = tryAcquire(taskId, 1); if (intercepted) { //被限制的次数累积 limited.getAndIncrement(); } Thread.sleep(200); } } catch (Exception e) { e.printStackTrace(); } //等待所有线程结束 countDownLatch.countDown(); }); } try { countDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } float time = (System.currentTimeMillis() - start) / F; //输出统计结果 log.info("限制的次数为：" + limited.get() + ",通过的次数为：" + (threads *turns - limited.get()));限制的比例为 log.info("限制的比例为：" +(float) limited.get() / (float) (threads *turns)); log.info("运行的时长为：" + time); }}

运行这个示例程序，部分结果如下：

[pool-2-thread-2] INFO c.c.s.r.TokenBucketLimiter - tokens 0 capacity 2 gap 104[pool-2-thread-1] INFO c.c.s.r.TokenBucketLimiter - tokens 0 capacity 2 gap 114[pool-2-thread-2] INFO c.c.s.r.TokenBucketLimiter - tokens 0 capacity 2 gap 314[pool-2-thread-1] INFO c.c.s.r.TokenBucketLimiter - tokens 0 capacity 2 gap 314[pool-2-thread-2] INFO c.c.s.r.TokenBucketLimiter - tokens 1 capacity 2 gap 515[pool-2-thread-1] INFO c.c.s.r.TokenBucketLimiter - tokens 0 capacity 2 gap 0...[pool-2-thread-2] INFO c.c.s.r.TokenBucketLimiter - tokens 0 capacity 2 gap 401[pool-2-thread-1] INFO c.c.s.r.TokenBucketLimiter - tokens 0 capacity 2 gap 402[main] INFO c.c.s.r.TokenBucketLimiter - 限制的次数为：34,通过的次数为：6[main] INFO c.c.s.r.TokenBucketLimiter - 限制的比例为：0.85[main] INFO c.c.s.r.TokenBucketLimiter - 运行的时长为：4.119

令牌桶的好处之一就是可以方便地应对突发流量。比如，可以改变令牌的发放速度，算法能按照新的发送速率调大令牌的发放数量，使得突发流量能被处理。