Redis 实现限流策略
除了控制流量，限流还有一个应用目的是用于控制用户行为，避免垃圾请求。对非法行为，业务必须规定适当的惩处策略。
除了控制流量，限流还有一个应用目的是用于控制用户行为，避免垃圾请求。
比如在 UGC 社区，用户的发帖、回复、点赞等行为都要严格受控，一般要严格限定某行为在规定 时间内允许的次数，超过了次数那就是非法行为。对非法行为，业务必须规定适当的惩处策略。
如何使用 Redis 来实现简单限流策略？
接口的定义
# 指定用户 user_id 的某个行为 action_key 在特定的时间内 period 只允许发生一定的次数
max_count
def is_action_allowed(user_id, action_key, period, max_count):
return True
# 调用这个接口 , 一分钟内只允许最多回复 5 个帖子
can_reply = is_action_allowed("laoqian", "reply", 60, 5)
if can_reply:
do_reply()
else:
raise ActionThresholdOverflow()解决方案
这个限流需求中存在一个滑动时间窗口，想想 zset 数据结构的 score 值，是不是可以通过 score 来圈出这个时间窗口来。
而且我们只需要保留这个时间窗口，窗口之外的数据都可以砍掉。那这个 zset 的 value 填什么比较合适呢？它只需要保证唯一性即可，用 uuid 会比较浪费空间，那就改用毫秒时间戳吧。
如图所示，用一个 zset 结构记录用户的行为历史，每一个行为都会作为 zset 中的一个 key 保存下来。同一个用户同一种行为用一个 zset 记录。
为节省内存，我们只需要保留时间窗口内的行为记录，同时如果用户是冷用户，滑动时间窗口内的行为是空记录，那么这个 zset 就可以从内存中移除，不再占用空间。
通过统计滑动窗口内的行为数量与阈值 max_count 进行比较就可以得出当前的行为是否 允许。用代码表示如下：
import org.apache.shiro.util.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
import java.util.List;
@Component
public class SimpleRateLimiter {
@Resource
private RedisTemplate<String, Object> redisTemplate;
public boolean isActionAllowed(String userId, String actionKey, int period, long maxCount) {
String key = String.format("hist:%s:%s", userId, actionKey);
long nowMills = System.currentTimeMillis();
List<Object> objects = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
// 打开管道
connection.openPipeline();
byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
// 添加命令
connection.zAdd(keyBytes, nowMills
, String.valueOf(nowMills).getBytes(StandardCharsets.UTF_8));
// 清楚无用数据
connection.zRemRangeByScore(keyBytes, 0, nowMills - period * 1000L);
// 判断规定时间内请求数量
connection.zCard(keyBytes);
// 重新设置过期时间
connection.expire(keyBytes, period + 1);
// 关闭管道 不需要close 否则拿不到返回值
// connection.closePipeline();
// 这里一定要返回null，最终pipeline的执行结果，才会返回给最外层
return null;
});
if (!CollectionUtils.isEmpty(objects)) {
return (Long) objects.get(2) <= maxCount;
}
return true;
}
}
测试：
import com.hcx.common.redisdemo.SimpleRateLimiter;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class MallCommonApplicationTests {
@Autowired
private SimpleRateLimiter limiter;
@Test
void contextLoads() {
for (int i = 0; i < 20; i++) {
System.out.println("结果:" + limiter.isActionAllowed("laoqian", "reply", 60, 5));
}
}
}
/*
结果:true
结果:true
结果:true
结果:true
结果:true
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
结果:false
*/
整体思路就是：每一个行为到来时，都维护一次时间窗口。将时间窗口外的记录全部清理掉，只保留窗口内的记录。
zset 集合中只有 score 值非常重要，value 值没有特别的意义，只需要保证它是唯一的就可以了。 因为这几个连续的 Redis 操作都是针对同一个 key 的，使用 pipeline 可以显著提升 Redis 存取效率。
但这种方案也有缺点，因为它要记录时间窗口内所有的行为记录，如果这个量很大，比如限定 60s 内操作不得超过 100w 次这样的参数，它是不适合做这样的限流的，因为会消耗大量的存储空间。
高级限流算法——漏斗限流
漏斗限流是最常用的限流方法之一，顾名思义，这个算法的灵感源于漏斗（funnel）的结构。
漏洞的容量是有限的，如果将漏嘴堵住，然后一直往里面灌水，它就会变满，直至再也 装不进去。如果将漏嘴放开，水就会往下流，流走一部分之后，就又可以继续往里面灌水。 如果漏嘴流水的速率大于灌水的速率，那么漏斗永远都装不满。
如果漏嘴流水速率小于灌水的速率，那么一旦漏斗满了，灌水就需要暂停并等待漏斗腾空。
所以，漏斗的剩余空间就代表着当前行为可以持续进行的数量，漏嘴的流水速率代表着 系统允许该行为的最大频率。
单机版漏斗算法实现_JAVA
import java.util.HashMap;
import java.util.Map;
public class FunnelRateLimiter {
private final Map<String, Funnel> funnels = new HashMap<>();
public static void main(String[] args) {
FunnelRateLimiter limiter = new FunnelRateLimiter();
for (int i = 0; i < 20; i++) {
System.out.println("次数" + (i + 1) + ": " + limiter.isActionAllowed("user", "get", 10, 1));
}
}
/**
* @param userId 用户ID
* @param actionKey 行为key
* @param capacity 漏斗容量
* @param leakingRate 漏嘴流水速率 单位mill
* @return 行为能否执行
*/
public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate) {
String key = String.format("%s:%s", userId, actionKey);
Funnel funnel = funnels.get(key);
if (funnel == null) {
funnel = new Funnel(capacity, leakingRate);
funnels.put(key, funnel);
}
return funnel.watering(1); // 需要 1 个 quota
}
static class Funnel {
/**
* 漏斗容量
*/
int capacity;
/**
* 漏嘴流水速率
*/
float leakingRate;
/**
* 漏斗剩余空间
*/
int leftQuota;
/**
* 上一次漏水时间
*/
long leakingTs;
public Funnel(int capacity, float leakingRate) {
this.capacity = capacity;
this.leakingRate = leakingRate;
this.leftQuota = capacity;
this.leakingTs = System.currentTimeMillis();
}
/**
* 计算剩余空间
*/
void makeSpace() {
long nowTs = System.currentTimeMillis();
long deltaTs = nowTs - leakingTs; // 距离上一次漏水过去了多久
int deltaQuota = (int) (deltaTs * leakingRate); // 腾出多少空间
if (deltaQuota < 0) { // 间隔时间太长，整数数字过大溢出
this.leftQuota = capacity;
this.leakingTs = nowTs;
return;
}
if (deltaQuota < 1) { // 腾出空间太小，最小单位是 1
return;
}
this.leftQuota += deltaQuota;
this.leakingTs = nowTs;
if (this.leftQuota > this.capacity) {
this.leftQuota = this.capacity;
}
}
/**
* @param quota 待申请空间
* @return 申请是否成功
*/
boolean watering(int quota) {
makeSpace();
if (this.leftQuota >= quota) {
this.leftQuota -= quota;
return true;
}
return false;
}
}
}
/*
次数1: true
次数2: true
次数3: true
次数4: true
次数5: true
次数6: true
次数7: true
次数8: true
次数9: true
次数10: true
次数11: false
次数12: false
次数13: false
次数14: false
次数15: false
次数16: false
次数17: false
次数18: false
次数19: false
次数20: false
*/
Funnel 对象的 make_space 方法是漏斗算法的核心，其在每次灌水前都会被调用以触发漏水，给漏斗腾出空间来。
能腾出多少空间取决于过去了多久以及流水的速率。Funnel 对象占据的空间大小不再和行为的频率成正比，它的空间占用是一个常量。
分布式的漏斗算法该如何实现？能不能使用 Redis 的基础数据结构来搞定？
观察 Funnel 对象的几个字段，发现可以将 Funnel 对象的内容按字段存储到一 个 hash 结构中，灌水的时候将 hash 结构的字段取出来进行逻辑运算后，再将新值回填到 hash 结构中就完成了一次行为频度的检测。
但是有个问题，无法保证整个过程的原子性。从 hash 结构中取值，然后在内存里运算，再回填到 hash 结构，这三个过程无法原子化，意味着需要进行适当的加锁控制。
而一旦加锁，就意味着会有加锁失败，加锁失败就需要选择重试或者放弃。
- 如果重试的话，就会导致性能下降。
- 如果放弃的话，就会影响用户体验。
同时，代码的复杂度也跟着升高很多。
Redis-Cell——限流 Redis 模块
Redis 4.0 提供了一个限流 Redis 模块，它叫 redis-cell。该模块也使用了漏斗算法，并 提供了原子的限流指令。
该模块只有 1 条指令 cl.throttle，它的参数和返回值都略显复杂，接下来让我们来看看这 个指令具体该如何使用。
上面这个指令的意思是允许「用户老钱回复行为」的频率为每 60s 最多 30 次(漏水速 率)，漏斗的初始容量为 16，也就是说一开始可以连续回复 16 个帖子，然后才开始受漏水速率的影响。我们看到这个指令中漏水速率变成了 2 个参数，替代了之前的单个浮点数。用 两个参数相除的结果来表达漏水速率相对单个浮点数要更加直观一些。
- > cl.throttle laoqian:reply 15 30 60
- 1) (integer) 0 # 0 表示允许，1 表示拒绝
- 2) (integer) 15 # 漏斗容量 capacity
- 3) (integer) 14 # 漏斗剩余空间 left_quota
- 4) (integer) -1 # 如果拒绝了，需要多长时间后再试(漏斗有空间了，单位秒)
- 5) (integer) 2 # 多长时间后，漏斗完全空出来(left_quota==capacity，单位秒)
在执行限流指令时，如果被拒绝了，就需要丢弃或重试。cl.throttle 指令考虑的非常周 到，连重试时间都帮你算好了，直接取返回结果数组的第四个值进行 sleep 即可，如果不想 阻塞线程，也可以异步定时任务来重试。
package com.hcx.common.redisdemo;
import org.apache.shiro.util.CollectionUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Collections;
import java.util.List;
@Component
public class FunnelRateLimiter {
/**
* lua 脚本
*/
public static final String LUA_SCRIPT = "return redis.call('cl.throttle',KEYS[1], ARGV[1], ARGV[2], ARGV[3], ARGV[4])";
@Resource
private RedisTemplate<String, Object> redisTemplate;
/**
* @param key 键值
* @param capacity 总漏斗容量-1
* @param operations 漏水速率
* @param seconds 漏水周期
* @param quota 待申请空间
* @return 申请结果
*/
public boolean isActionAllowed(String key, int capacity, int operations, int seconds, int quota) {
try {
DefaultRedisScript<List> script = new DefaultRedisScript<>(LUA_SCRIPT, List.class);
List<Long> rs = redisTemplate.execute(script, Collections.singletonList(key), capacity, operations, seconds, quota);
if (CollectionUtils.isEmpty(rs)) return false;
System.out.println("漏斗容量：" + rs.get(1));
System.out.println("剩余空间：" + rs.get(2));
System.out.println("最少多长时间后再试：" + rs.get(3));
System.out.println("多长时间后漏斗为空：" + rs.get(4));
return rs.get(0) == 0;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
}
/*
漏斗容量：16
剩余空间：15
最少多长时间后再试：-1
多长时间后漏斗为空：2
结果:true
……
最少多长时间后再试：-1
多长时间后漏斗为空：27
结果:true
漏斗容量：16
剩余空间：1
最少多长时间后再试：-1
多长时间后漏斗为空：29
结果:true
漏斗容量：16
剩余空间：0
最少多长时间后再试：-1
多长时间后漏斗为空：31
结果:true
漏斗容量：16
剩余空间：0
最少多长时间后再试：1
多长时间后漏斗为空：31
结果:false
漏斗容量：16
剩余空间：0
最少多长时间后再试：1
多长时间后漏斗为空：31
结果:false
漏斗容量：16
剩余空间：0
最少多长时间后再试：1
多长时间后漏斗为空：31
结果:false
漏斗容量：16
剩余空间：0
最少多长时间后再试：1
多长时间后漏斗为空：31
结果:false
*/
在执行限流指令时，如果被拒绝了，就需要丢弃或重试。cl.throttle 指令考虑的非常周 到，连重试时间都帮你算好了，直接取返回结果数组的第四个值进行 sleep 即可，如果不想 阻塞线程，也可以异步定时任务来重试。
为开发者提供学习成长、分享交流、生态实践、资源工具等服务，帮助开发者快速成长。
更多推荐
1
0- 0
已为社区贡献1条内容
所有评论(0)