Redis实现限流的几种方式

参考文章：https://zhuanlan.zhihu.com/p/439093222

https://mp.weixin.qq.com/s/zf9uqfJfRYvmSVXUQofF2A

互联网应用往往是高并发的场景，互联网的特性就是瞬时、激增，比如鹿晗官宣了，此时，如果没有流量管控，很容易导致系统雪崩。

而限流是用来保证系统稳定性的常用手段，当系统遭遇瞬时流量激增，很可能会因系统资源耗尽导致宕机，限流可以把一超出系统承受能力外的流量直接拒绝掉，保证大部分流量可以正常访问，从而保证系统只接收承受范围以内的请求。

我们常用的限流算法有：漏桶算法、令牌桶算法。

漏桶算法

漏桶算法很形象，我们可以想像有一个大桶，大桶底部有一个固定大小的洞，Web请求就像水一样，先进入大桶，然后以固定的速率从底部漏出来，无论进入桶中的水多么迅猛，漏桶算法始终以固定的速度来漏水。

对应到Web请求就是：

• 当桶中无水时表示当前无请求等待，可以直接处理当前的请求；


• 当桶中有水时表示当前有请求正在等待处理，此时新来的请求也是需要进行等待处理；


• 当桶中水已经装满，并且进入的速率大于漏水的速率，水就会溢出来，此时系统就会拒绝新来的请求；

令牌桶算法

令牌桶跟漏桶算法有点不一样，令牌桶算法也有一个大桶，桶中装的都是令牌，有一个固定的“人”在不停的往桶中放令牌，每个请求来的时候都要从桶中拿到令牌，要不然就无法进行请求操作。

• 当没有请求来时，桶中的令牌会越来越多，一直到桶被令牌装满为止，多余的令牌会被丢弃


• 当请求的速率大于令牌放入桶的速率，桶中的令牌会越来越少，直止桶变空为止，此时的请求会等待新令牌的产生

漏桶算法 VS 令牌桶算法

• 漏桶算法是桶中有水就需要等待，桶满就拒绝请求。而令牌桶是桶变空了需要等待令牌产生;


• 漏桶算法漏水的速率固定，令牌桶算法往桶中放令牌的速率固定;


• 令牌桶可以接收的瞬时流量比漏桶大，比如桶的容量为100，令牌桶会装满100个令牌，当有瞬时80个并发过来时可以从桶中迅速拿到令牌进行处理，而漏桶的消费速率固定，当瞬时80个并发过来时，可能需要进行排队等待;

介绍了算法，接下来我们介绍下Redis实现限流的几种方式。

第一种：基于Redis的setNX的操作

我们在使用Redis的分布式锁的时候，大家都知道是依靠了setNX的指令，在CAS（Compare and swap）的操作的时候，同时给指定的key设置了过期实践（expire），我们在限流的主要目的就是为了在单位时间内，有且仅有N数量的请求能够访问我的代码程序。所以依靠setnx可以很轻松的做到这方面的功能。

比如我们需要在10秒内限定20个请求，那么我们在setnx的时候可以设置过期时间10，当请求的setnx数量达到20时候即达到了限流效果。代码比较简单就不做展示了。

当然这种做法的弊端是很多的，比如当统计1-10秒的时候，无法统计2-11秒之内，如果需要统计N秒内的M个请求，那么我们的Redis中需要保持N个key等等问题。

第二种：基于Redis的数据结构zset

其实限流涉及的最主要的就是滑动窗口，上面也提到1-10怎么变成2-11。其实也就是起始值和末端值都各+1即可。

而我们如果用Redis的list数据结构可以轻而易举的实现该功能

我们可以将请求打造成一个zset数组，当每一次请求进来的时候，value保持唯一，可以用UUID生成，而score可以用当前时间戳表示，因为score我们可以用来计算当前时间戳之内有多少的请求数量。而zset数据结构也提供了range方法让我们可以很轻易的获取到2个时间戳内有多少请求（解决了第一种方案中无法统计2-11秒的问题）。

示例代码：

using System;
using System.Threading;
using ServiceStack.Redis;
namespace IPCounter
{
class Program
{
static void Main(string[] args)
{
RedisClient client = new RedisClient("1633com@192.168.1.128:6379");
string key = "aa";
for (int i=0;i<100;i++)
{
long currentTime = ToUnixTimestampBySeconds(DateTime.Now);

if (client.ContainsKey(key))
{
var count = client.GetRangeFromSortedSetByHighestScore(key,currentTime-1,currentTime).Count;
if(count>3)
{

Console.WriteLine("您的请求频率太高了");
Console.ReadLine();
}
count = client.GetRangeFromSortedSetByHighestScore(key, currentTime - 60, currentTime).Count;
if (count > 30)
{
Console.WriteLine("您的请求频率太高了");
Console.ReadLine();
}
}
string value = Guid.NewGuid().ToString();
long score = currentTime;
client.AddItemToSortedSet(key, value, score);
//清除2分钟之前的记录
var list = client.GetRangeFromSortedSetByHighestScore(key,0, currentTime-65);
client.RemoveItemsFromSortedSet(key, list);
Thread.Sleep(500);
}
Console.WriteLine("Hello World!");
}
public static long ToUnixTimestampBySeconds(DateTime dt)
{
DateTimeOffset dto = new DateTimeOffset(dt);
return dto.ToUnixTimeSeconds();
}
}
}

通过上述代码可以做到滑动窗口的效果，并且能保证每N秒内至多M个请求，缺点就是zset的数据结构会越来越大。实现方式相对也是比较简单的。

第三种：基于Redis的令牌桶算法

令牌桶算法提及到输入速率和输出速率，当输出速率大于输入速率，那么就是超出流量限制了。

也就是说我们每访问一次请求的时候，可以从Redis中获取一个令牌，如果拿到令牌了，那就说明没超出限制，而如果拿不到，则结果相反。

依靠上述的思想，我们可以结合Redis的List数据结构很轻易的做到这样的代码，只是简单实现依靠List的leftPop方法来获取令牌。

示例代码：

// 输出令牌

static void LimitRequest()
{
RedisClient client = new RedisClient("1633com@192.168.1.128:6379");
string key = "limitRate";
var result= client.PopItemFromList(key);
if(result==null)
{
Console.WriteLine("系统繁忙，请稍后再试");
}
else
{
Console.WriteLine("访问成功");
}
}

再依靠定时任务，定时往令牌桶List中加入新的令牌（使用List的rightPush方法），当然令牌也需要唯一性，这里还是用UUID生成令牌：

// 10S的速率往令牌桶中添加UUID，保证唯一性

///

综上，代码实现起始都不是很难，针对这些限流方式我们可以在AOP或者filter中加入以上代码，用来做到接口的限流，最终保护系统的稳定。