Redis 實現多規則限流的思考與實踐
作者：張張
市面上很多介紹redis如何實現限流的，但是大部分都有一個缺點，就是只能實現單一的限流，比如1分鐘訪問1次或者60分鐘訪問10次這種，但是如果想一個接口兩種規則都需要滿足呢，我們的項目又是分布式項目，應該如何解決，下面就介紹一下redis實現分布式多規則限流的方式。
簡介
市面上很多介紹redis如何實現限流的，但是大部分都有一個缺點，就是只能實現單一的限流，比如1分鐘訪問1次或者60分鐘訪問10次這種，但是如果想一個接口兩種規則都需要滿足呢，我們的項目又是分布式項目，應該如何解決，下面就介紹一下redis實現分布式多規則限流的方式。
思考
- 如何一分鐘只能發送一次驗證碼，一小時只能發送10次驗證碼等等多種規則的限流
- 如何防止接口被惡意打擊（短時間內大量請求）
- 如何限制接口規定時間內訪問次數
解決方法
記錄某IP訪問次數
使用 String結構 記錄固定時間段內某用戶IP訪問某接口的次數
- RedisKey = prefix : className : methodName
- RedisVlue = 訪問次數
攔截請求:
- 初次訪問時設置 「[RedisKey] [RedisValue=1] [規定的過期時間]」
- 獲取 RedisValue 是否超過規定次數，超過則攔截，未超過則對 RedisKey 進行加1
分析: 規則是每分鐘訪問 1000 次
- 考慮并發問題
假設目前 RedisKey => RedisValue 為 999
目前大量請求進行到第一步( 獲取Redis請求次數 )，那么所有線程都獲取到了值為999，進行判斷都未超過限定次數則不攔截，導致實際次數超過 1000 次
「解決辦法:」 保證方法執行原子性（加鎖、lua）
- 考慮在臨界值進行訪問
思考下圖
圖片
代碼實現: 比較簡單
參考：https://gitee.com/y_project/RuoYi-Vue/blob/master/ruoyi-framework/src/main/java/com/ruoyi/framework/aspectj/RateLimiterAspect.java。
Zset解決臨界值問題
使用 Zset 進行存儲，解決臨界值訪問問題
圖片
網上幾乎都有實現，這里就不過多介紹
實現多規則限流
先確定最終需要的效果
- 能實現多種限流規則
- 能實現防重復提交
通過以上要求設計注解（先想象出最終實現效果）
@RateLimiter(
rules = {
// 60秒內只能訪問10次
@RateRule(count = 10, time = 60, timeUnit = TimeUnit.SECONDS),
// 120秒內只能訪問20次
@RateRule(count = 20, time = 120, timeUnit = TimeUnit.SECONDS)
},
// 防重復提交 （5秒鐘只能訪問1次）
preventDuplicate = true
)編寫注解（RateLimiter，RateRule）
編寫 RateLimiter 注解。
/**
* @Description: 請求接口限制
* @Author: yiFei
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface RateLimiter {
/**
* 限流key
*/
String key() default RedisKeyConstants.RATE_LIMIT_CACHE_PREFIX;
/**
* 限流類型 ( 默認 Ip 模式 )
*/
LimitTypeEnum limitType() default LimitTypeEnum.IP;
/**
* 錯誤提示
*/
ResultCode message() default ResultCode.REQUEST_MORE_ERROR;
/**
* 限流規則 （規則不可變，可多規則）
*/
RateRule[] rules() default {};
/**
* 防重復提交值
*/
boolean preventDuplicate() default false;
/**
* 防重復提交默認值
*/
RateRule preventDuplicateRule() default @RateRule(count = 1, time = 5);
}編寫RateRule注解
@Target(ElementType.ANNOTATION_TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface RateRule {
/**
* 限流次數
*/
long count() default 10;
/**
* 限流時間
*/
long time() default 60;
/**
* 限流時間單位
*/
TimeUnit timeUnit() default TimeUnit.SECONDS;
}攔截注解 RateLimiter
- 確定redis存儲方式
RedisKey = prefix : className : methodName
RedisScore = 時間戳
RedisValue = 任意分布式不重復的值即可
- 編寫生成 RedisKey 的方法
/**
* 通過 rateLimiter 和 joinPoint 拼接 prefix : ip / userId : classSimpleName - methodName
*
* @param rateLimiter 提供 prefix
* @param joinPoint 提供 classSimpleName : methodName
* @return
*/
public String getCombineKey(RateLimiter rateLimiter, JoinPoint joinPoint) {
StringBuffer key = new StringBuffer(rateLimiter.key());
// 不同限流類型使用不同的前綴
switch (rateLimiter.limitType()) {
// XXX 可以新增通過參數指定參數進行限流
case IP:
key.append(IpUtil.getIpAddr(((ServletRequestAttributes) Objects.requireNonNull(RequestContextHolder.getRequestAttributes())).getRequest())).append(":");
break;
case USER_ID:
SysUserDetails user = SecurityUtil.getUser();
if (!ObjectUtils.isEmpty(user)) key.append(user.getUserId()).append(":");
break;
case GLOBAL:
break;
}
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
Class<?> targetClass = method.getDeclaringClass();
key.append(targetClass.getSimpleName()).append("-").append(method.getName());
return key.toString();
}編寫lua腳本
編寫lua腳本 （兩種將時間添加到Redis的方法）。
Zset的UUID value值
UUID（可用其他有相同的特性的值）為Zset中的value值
- 參數介紹
KEYS[1] = prefix : ? : className : methodName
KEYS[2] = 唯一ID
KEYS[3] = 當前時間
ARGV = [次數，單位時間，次數，單位時間, 次數, 單位時間 ...]
- 由java傳入分布式不重復的 value 值
-- 1. 獲取參數
local key = KEYS[1]
local uuid = KEYS[2]
local currentTime = tonumber(KEYS[3])
-- 2. 以數組最大值為 ttl 最大值
local expireTime = -1;
-- 3. 遍歷數組查看是否超過限流規則
for i = 1, #ARGV, 2 do
local rateRuleCount = tonumber(ARGV[i])
local rateRuleTime = tonumber(ARGV[i + 1])
-- 3.1 判斷在單位時間內訪問次數
local count = redis.call('ZCOUNT', key, currentTime - rateRuleTime, currentTime)
-- 3.2 判斷是否超過規定次數
if tonumber(count) >= rateRuleCount then
return true
end
-- 3.3 判斷元素最大值，設置為最終過期時間
if rateRuleTime > expireTime then
expireTime = rateRuleTime
end
end
-- 4. redis 中添加當前時間
redis.call('ZADD', key, currentTime, uuid)
-- 5. 更新緩存過期時間
redis.call('PEXPIRE', key, expireTime)
-- 6. 刪除最大時間限度之前的數據，防止數據過多
redis.call('ZREMRANGEBYSCORE', key, 0, currentTime - expireTime)
return false根據時間戳作為Zset中的value值
- 參數介紹
KEYS[1] = prefix : ? : className : methodName
KEYS[2] = 當前時間
ARGV = [次數，單位時間，次數，單位時間, 次數, 單位時間 ...]
- 根據時間進行生成value值，考慮同一毫秒添加相同時間值問題
以下為第二種實現方式，在并發高的情況下效率低，value是通過時間戳進行添加，但是訪問量大的話會使得一直在調用 redis.call('ZADD', key, currentTime, currentTime)，但是在不沖突value的情況下，會比生成 UUID 好
-- 1. 獲取參數
local key = KEYS[1]
local currentTime = KEYS[2]
-- 2. 以數組最大值為 ttl 最大值
local expireTime = -1;
-- 3. 遍歷數組查看是否越界
for i = 1, #ARGV, 2 do
local rateRuleCount = tonumber(ARGV[i])
local rateRuleTime = tonumber(ARGV[i + 1])
-- 3.1 判斷在單位時間內訪問次數
local count = redis.call('ZCOUNT', key, currentTime - rateRuleTime, currentTime)
-- 3.2 判斷是否超過規定次數
if tonumber(count) >= rateRuleCount then
return true
end
-- 3.3 判斷元素最大值，設置為最終過期時間
if rateRuleTime > expireTime then
expireTime = rateRuleTime
end
end
-- 4. 更新緩存過期時間
redis.call('PEXPIRE', key, expireTime)
-- 5. 刪除最大時間限度之前的數據，防止數據過多
redis.call('ZREMRANGEBYSCORE', key, 0, currentTime - expireTime)
-- 6. redis 中添加當前時間 ( 解決多個線程在同一毫秒添加相同 value 導致 Redis 漏記的問題 )
-- 6.1 maxRetries 最大重試次數 retries 重試次數
local maxRetries = 5
local retries = 0
while true do
local result = redis.call('ZADD', key, currentTime, currentTime)
if result == 1 then
-- 6.2 添加成功則跳出循環
break
else
-- 6.3 未添加成功則 value + 1 再次進行嘗試
retries = retries + 1
if retries >= maxRetries then
-- 6.4 超過最大嘗試次數 采用添加隨機數策略
local random_value = math.random(1, 1000)
currentTime = currentTime + random_value
else
currentTime = currentTime + 1
end
end
end
return false編寫 AOP 攔截
@Autowired
private RedisTemplate<String, Object> redisTemplate;
@Autowired
private RedisScript<Boolean> limitScript;
/**
* 限流
* XXX 對限流要求比較高，可以使用在 Redis中對規則進行存儲校驗 或者使用中間件
*
* @param joinPoint joinPoint
* @param rateLimiter 限流注解
*/
@Before(value = "@annotation(rateLimiter)")
public void boBefore(JoinPoint joinPoint, RateLimiter rateLimiter) {
// 1. 生成 key
String key = getCombineKey(rateLimiter, joinPoint);
try {
// 2. 執行腳本返回是否限流
Boolean flag = redisTemplate.execute(limitScript,
ListUtil.of(key, String.valueOf(System.currentTimeMillis())),
(Object[]) getRules(rateLimiter));
// 3. 判斷是否限流
if (Boolean.TRUE.equals(flag)) {
log.error("ip: '{}' 攔截到一個請求 RedisKey: '{}'",
IpUtil.getIpAddr(((ServletRequestAttributes) Objects.requireNonNull(RequestContextHolder.getRequestAttributes())).getRequest()),
key);
throw new ServiceException(rateLimiter.message());
}
} catch (ServiceException e) {
throw e;
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 獲取規則
*
* @param rateLimiter 獲取其中規則信息
* @return
*/
private Long[] getRules(RateLimiter rateLimiter) {
int capacity = rateLimiter.rules().length << 1;
// 1. 構建 args
Long[] args = new Long[rateLimiter.preventDuplicate() ? capacity + 2 : capacity];
// 3. 記錄數組元素
int index = 0;
// 2. 判斷是否需要添加防重復提交到redis進行校驗
if (rateLimiter.preventDuplicate()) {
RateRule preventRateRule = rateLimiter.preventDuplicateRule();
args[index++] = preventRateRule.count();
args[index++] = preventRateRule.timeUnit().toMillis(preventRateRule.time());
}
RateRule[] rules = rateLimiter.rules();
for (RateRule rule : rules) {
args[index++] = rule.count();
args[index++] = rule.timeUnit().toMillis(rule.time());
}
return args;
}以上，歡迎大家提出意見。
責任編輯：武曉燕
來源：
架構精進之路
