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）

考虑在临界值进行访问

思考下图

代码实现: 比较简单，

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@interfaceRateLimiter{

/**
*限流key
*/
Stringkey()defaultRedisKeyConstants.RATE_LIMIT_CACHE_PREFIX;

/**
*限流类型(默认Ip模式)
*/
LimitTypeEnumlimitType()defaultLimitTypeEnum.IP;

/**
*错误提示
*/
ResultCodemessage()defaultResultCode.REQUEST_MORE_ERROR;

/**
*限流规则（规则不可变，可多规则）
*/
RateRule[]rules()default{};

/**
*防重复提交值
*/
booleanpreventDuplicate()defaultfalse;

/**
*防重复提交默认值
*/
RateRulepreventDuplicateRule()default@RateRule(count=1,time=5);
}

编写RateRule注解

@Target(ElementType.ANNOTATION_TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public@interfaceRateRule{

/**
*限流次数
*/
longcount()default10;

/**
*限流时间
*/
longtime()default60;

/**
*限流时间单位
*/
TimeUnittimeUnit()defaultTimeUnit.SECONDS;

}

拦截注解 RateLimiter

确定redis存储方式

RedisKey = prefix : className : methodName

RedisScore = 时间戳

RedisValue = 任意分布式不重复的值即可

编写生成 RedisKey 的方法

/**
*通过rateLimiter和joinPoint拼接prefix:ip/userId:classSimpleName-methodName
*
*@paramrateLimiter提供prefix
*@paramjoinPoint提供classSimpleName:methodName
*@return
*/
publicStringgetCombineKey(RateLimiterrateLimiter,JoinPointjoinPoint){
StringBufferkey=newStringBuffer(rateLimiter.key());
//不同限流类型使用不同的前缀
switch(rateLimiter.limitType()){
//XXX可以新增通过参数指定参数进行限流
caseIP:
key.append(IpUtil.getIpAddr(((ServletRequestAttributes)Objects.requireNonNull(RequestContextHolder.getRequestAttributes())).getRequest())).append(":");
break;
caseUSER_ID:
SysUserDetailsuser=SecurityUtil.getUser();
if(!ObjectUtils.isEmpty(user))key.append(user.getUserId()).append(":");
break;
caseGLOBAL:
break;
}
MethodSignaturesignature=(MethodSignature)joinPoint.getSignature();
Methodmethod=signature.getMethod();
ClasstargetClass=method.getDeclaringClass();
key.append(targetClass.getSimpleName()).append("-").append(method.getName());
returnkey.toString();
}

编写lua脚本

编写lua脚本 （两种将时间添加到Redis的方法）。

Zset的UUID value值

UUID（可用其他有相同的特性的值）为Zset中的value值

参数介绍

KEYS[1] = prefix : ? : className : methodName

KEYS[2] = 唯一ID

KEYS[3] = 当前时间

ARGV = [次数，单位时间，次数，单位时间, 次数, 单位时间 ...]

由java传入分布式不重复的 value 值

--1.获取参数
localkey=KEYS[1]
localuuid=KEYS[2]
localcurrentTime=tonumber(KEYS[3])
--2.以数组最大值为ttl最大值
localexpireTime=-1;
--3.遍历数组查看是否超过限流规则
fori=1,#ARGV,2do
localrateRuleCount=tonumber(ARGV[i])
localrateRuleTime=tonumber(ARGV[i+1])
--3.1判断在单位时间内访问次数
localcount=redis.call('ZCOUNT',key,currentTime-rateRuleTime,currentTime)
--3.2判断是否超过规定次数
iftonumber(count)>=rateRuleCountthen
returntrue
end
--3.3判断元素最大值，设置为最终过期时间
ifrateRuleTime>expireTimethen
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)
returnfalse

根据时间戳作为Zset中的value值

参数介绍

KEYS[1] = prefix : ? : className : methodName

KEYS[2] = 当前时间

ARGV = [次数，单位时间，次数，单位时间, 次数, 单位时间 ...]

根据时间进行生成value值，考虑同一毫秒添加相同时间值问题

以下为第二种实现方式，在并发高的情况下效率低，value是通过时间戳进行添加，但是访问量大的话会使得一直在调用 redis.call('ZADD', key, currentTime, currentTime)，但是在不冲突value的情况下，会比生成 UUID 好

--1.获取参数
localkey=KEYS[1]
localcurrentTime=KEYS[2]
--2.以数组最大值为ttl最大值
localexpireTime=-1;
--3.遍历数组查看是否越界
fori=1,#ARGV,2do
localrateRuleCount=tonumber(ARGV[i])
localrateRuleTime=tonumber(ARGV[i+1])
--3.1判断在单位时间内访问次数
localcount=redis.call('ZCOUNT',key,currentTime-rateRuleTime,currentTime)
--3.2判断是否超过规定次数
iftonumber(count)>=rateRuleCountthen
returntrue
end
--3.3判断元素最大值，设置为最终过期时间
ifrateRuleTime>expireTimethen
expireTime=rateRuleTime
end
end
--4.更新缓存过期时间
redis.call('PEXPIRE',key,expireTime)
--5.删除最大时间限度之前的数据，防止数据过多
redis.call('ZREMRANGEBYSCORE',key,0,currentTime-expireTime)
--6.redis中添加当前时间(解决多个线程在同一毫秒添加相同value导致Redis漏记的问题)
--6.1maxRetries最大重试次数retries重试次数
localmaxRetries=5
localretries=0
whiletruedo
localresult=redis.call('ZADD',key,currentTime,currentTime)
ifresult==1then
--6.2添加成功则跳出循环
break
else
--6.3未添加成功则value+1再次进行尝试
retries=retries+1
ifretries>=maxRetriesthen
--6.4超过最大尝试次数采用添加随机数策略
localrandom_value=math.random(1,1000)
currentTime=currentTime+random_value
else
currentTime=currentTime+1
end
end
end

returnfalse

编写 AOP 拦截

@Autowired
privateRedisTemplateredisTemplate;

@Autowired
privateRedisScriptlimitScript;

/**
*限流
*XXX对限流要求比较高，可以使用在Redis中对规则进行存储校验或者使用中间件
*
*@paramjoinPointjoinPoint
*@paramrateLimiter限流注解
*/
@Before(value="@annotation(rateLimiter)")
publicvoidboBefore(JoinPointjoinPoint,RateLimiterrateLimiter){
//1.生成key
Stringkey=getCombineKey(rateLimiter,joinPoint);
try{
//2.执行脚本返回是否限流
Booleanflag=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);
thrownewServiceException(rateLimiter.message());
}
}catch(ServiceExceptione){
throwe;
}catch(Exceptione){
e.printStackTrace();
}
}

/**
*获取规则
*
*@paramrateLimiter获取其中规则信息
*@return
*/
privateLong[]getRules(RateLimiterrateLimiter){
intcapacity=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;
}

审核编辑：刘清