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

限流原理与实战

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

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

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

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

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

限流策略原理与参考实现

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

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

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

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

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

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

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

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

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

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

package com.crazymaker.springcloud.ratelimit;
...
//计速器，限速
@Slf4j
public class CounterLimiter
{
 //起始时间
 private static long startTime = System.currentTimeMillis();
 //时间区间的时间间隔毫秒
 private static long interval = 1000;
 //每秒限制数量
 private static long maxCount = 2;
 //累加器
 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(0);
 startTime = nowTime;
 }
 }
 return 0;
 }
 }
 //线程池，用于多线程模拟测试
 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();
 long index = tryAcquire(taskId, j);
 if (index <= 0)
 {
 //被限制的次数累积
 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) / 1000F;
 //输出统计结果
 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
//漏桶限流
@Slf4j
public 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 / 1000;
 //剩余的水量 = 上次遗留的水量 - 漏出去的水
 water = water - outWater;
 log.info("water {} pastTime {} outWater {} ",
water, pastTime, outWater);
 //纠正剩余的水量
 if (water < 0)
 {
 water = 0;
 }
 //若剩余的水量小于等于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;
...
//令牌桶，限速
@Slf4j
public class TokenBucketLimiter
{
 //上一次令牌发放的时间
 public long lastTime = System.currentTimeMillis();
 //桶的容量
 public int capacity = 2;
 //令牌生成速度个/秒
 public int rate = 2;
 //当前令牌的数量
 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/ 1000));
 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) / 1000F;
 //输出统计结果
 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

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

本文给大家讲解的内容是高并发核心编程，限流原理与实战，限流策略原理与参考实现