基于redis实现分布式服务限流器
限流器概述
在后台开发中，服务端的限流器是一个很常见并且十分有用的组件，利用好限流器可以限制请求速率，保护后台服务。 比较常见的限流器分为两种，漏桶算法和令牌桶算法。
漏桶算法
漏桶算法原理很简单，用一个漏斗来控制请求的速率。在漏斗上方是收到的所有请求，请求就像水一样会进入漏斗中，同时漏斗也会以恒定的速度将水（请求）从下方进行排出，被排出的水（请求）才能访问服务。当请求量不大时候，如进水速率 < 出水速率那么其实漏斗并没有起到作用；当请求量很大的时候，超过漏斗容量的请求将被溢出，并且出水口可以一直保证恒定的速率。
令牌桶算法
令牌桶算法原理也很简单，假设我们的服务允许请求速度上限为5000次/分，那么这就意味着桶内的令牌数为5000，并且每隔一分钟桶内的令牌数就会被重置为5000。每一个请求过来都需要从桶内拿一块令牌，如果能取得令牌则允许访问服务，否则将会拒绝请求。
基于redis的分布式服务限流器
本文将基于redis来设计一个在分布式场景下的令牌桶算法，旨在重点解决以下问题：
- 并发请求如何处理？
- 何时进行加锁？何时不需要加锁？
- 如何提高准确性和稳定性？
流程图
详细设计
在实际场景中，服务的限流往往会和一些参数绑定在一起，比如：限制同一个ip地址的请求速率为5000次/分，限制某一个业务id的请求速率为5000次/分，根据这些绑定的变量数值，我们可以在redis中设置对应的key，通过不断累加该key对应的数值来实现限流器的设计。
计数器初始化
假设我们服务请求速率的最大值max为5000次/分。
当服务器收到请求时，首先判断redis中对应键k的数值v是否超过5000，如果是则拒绝请求，如果为否则继续判断v是否为0，当v为0的时候，我们需要进行初始化。初始化需要将v的值置为1，并且设置过期时间为60s。考虑以下几个问题：
- 初始化是否需要加锁？
- 为什么不能直接用
incr命令？
对于第一个问题，答案肯定是必然的，我们需要保证只有一个请求能进行初始化，否则在并发情况下会出现多个请求线程都对v进行置1操作，从而导致计数器不准确。
那么如何进行加锁操作呢？在分布式场景下是用本地锁是不正确的，因此我们同样可以利用redis的SET .. NX命令来实现分布式锁，来保证只有一个线程能进行初始化。
有一个需要注意的细节是：线程在获得锁之后，还需要在读取一次v的值，如果此时读取到数值不为0则说明在此之前已经被其他线程捷足先登了，此时就应该放弃初始化。
对于第二个问题，虽然redis的incr命令也可以保证只有一个请求线程能进行置1操作（因为redis是单线程的，天然满足锁），但是incr没有办法设置过期时间，因此不能直接使用incr命令。
代码如下：
func initCount(key string) (int, error) {
lockKey := key + "_" + initLockKeySuffix
getLock := false
var err error
// 一个循环去抢占锁
for !getLock {
// 再读取一次count数值，保证只有一个线程进行初始化
// 这一步很重要，可能在抢锁的过程中已经有其他线程完成了初始化，那么此线程就不需要初始化了
count, err := redis.GetInt(key)
if err != nil {
// 记录错误继续抢占初始化
log.Errorf("get redis failed, err: %v", err)
time.Sleep(time.Millisecond * 10)
continue
}
if count > 0 {
// 不是第一个线程，放弃初始化
return -1, nil
}
// 设置一个3s过期的nx锁
getLock, err = redis.SetNxWithExpire(lockKey, "ok", 3)
if err != nil {
// 记录错误继续抢占初始化
log.Errorf("set nx failed, err: %v", err)
time.Sleep(time.Millisecond * 10)
continue
}
if getLock {
// 抢到，退出循环
break
} else {
// 没抢到锁，等下一次
time.Sleep(time.Millisecond * 100)
}
}
// 获得锁之后开始进行初始化
ok, err := redis.SetIntWithExpire(key, 1, comm.Interval)
if err != nil || !ok {
// 初始化失败
return 0, myError.WithMessage(err, "redis init failed")
}
log.Info("init success")
// 删除锁
e := redis.DelKey(lockKey)
if e != nil {
// 只能被动等待锁过期
log.Errorf("redis del key failed, key: %s, err: %s", key, e)
}
// 初始化成功
return 1, nil
}
请求次数累加
对于上述抢占失败的线程，以及新来的请求线程就没有必要继续初始化了，而是直接对v值进行加1操作。考虑以下几个问题：
- 加
1操作是否需要加锁？ - 如何解决边界问题？
- 如何提升程序效率？
使用redis的incr命令进行加1操作，由于redis天然是单线程的，因此加1操作是不需要进行加锁的。对于每一个请求，可以通过判断incr返回值是否大于max来决定是否拒绝请求。
在并发的情况下，假设我们的服务限制访问速率为5000次/分，在某一时刻t请求数量已经达到了4999次，此时突然并发来了10个请求，按照上面设计的流程，这10个请求首先读取redis中对应键k的数值v，同时读取到了4999这个值，那么则会都进行加1操作，于是redis中对应键k最终值则为5009，超过了5000，虽然不影响服务，但是redis中值却超过了预期值，为了解决边界问题，我采用了阈值法，根据业务的需求可以事先估计一个阈值δ，比如80%，当redis中对应键k的数值v小于max * δ时，则不加锁直接使用incr进行加1，当超过时，则进行加锁排队加1。
代码如下
func increaseCountWithoutLock(key string) (int, error) {
// 直接进行加1，上层对加1后的数值进行判断
return redis.IncrOne(key)
}
func increaseCountWithLock(bid int, key string, count int) (int, error) {
if float64(count) < comm.RateThreshold*comm.RateLimit {
// 没有达到阈值，直接使用redis的incr来保证原子性
return redis.IncrOne(key)
}
// 达到阈值后incr操作需要排队
newCount, err := increaseSerialized(bid, key)
if err != nil {
return 0, myError.WithMessage(err, "increaseSerialized failed")
}
return newCount, nil
}
func increaseSerialized(bid int, key string) (int, error) {
lockKey := strconv.Itoa(bid) + incrLockKeySuffix
getLock := false
var err error
for !getLock {
// 再读取一次
oldCount, err := redis.GetInt(key)
if err != nil {
// 记录错误继续抢占资源
log.Errorf("set nx failed, err: %v", err)
time.Sleep(time.Millisecond * 10)
continue
}
if oldCount >= comm.RateLimit {
// redis不用加1，直接返回
return oldCount + 1, nil
}
// 设置一个3s过期的nx锁
getLock, err = redis.SetNxWithExpire(lockKey, "ok", 3)
if err != nil {
return -1, myError.WithMessage(err, "set redis lock failed")
}
if getLock {
break
} else {
time.Sleep(time.Millisecond * 100)
}
}
// 获得锁之后开始进行加1操作
newCount, err := redis.IncrOne(key)
// 删除锁
e := redis.DelKey(lockKey)
if e != nil {
// 只能被动等待锁过期
log.Errorf("redis del key failed, key: %s, err: %s", key, e)
}
return newCount, err
}
第一种方式不需要加锁，代码简单，但是没有保证redis中计数器的正确性，即没有满足解决问题（但是不影响业务）；第二种方式在达到阈值后需要加锁，代码较为复杂。
重置计数器
在初始化redis计数器时，我们使用了SET...EX方式设置了过期时间，但是在实际中可能出现key过期后却没有自动删除的现象，于是这里加上了手动删除过期key的监控，采用redis的ttl和del命令组合来重置计数器。
代码如下
func ttlCount(key string) error {
leftTime, err := redis.GetTtl(key)
log.Infof("key: %s, left time: %d", key, leftTime)
if err != nil {
return myError.WithMessage(err, "")
}
if leftTime == -1 || leftTime > comm.Interval {
// 说明此时key没有设置过期时间或者超时时间出错，则进行删除
return redis.DelKey(key)
}
return nil
}