限流原理与实战
在通信领域中，限流技术（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）漏桶容量是不变的，如果处理速度太慢，桶内水量会超出桶的容量，后面流入的水就会溢出，表示请求拒绝。
漏桶原理示意图
漏桶的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）如果令牌的发放速度慢于请求到来的速度，桶内就无令牌可领，请求就会被拒绝。
令牌桶
令牌桶的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
令牌桶的好处之一就是可以方便地应对突发流量。比如，可以改变令牌的发放速度，算法能按照新的发送速率调大令牌的发放数量，使得突发流量能被处理。
网友评论