基于Redis的分布式限流器Java实现

常见的限流方案

从实现方式上来讲，限流可分为简单计数器限流、滑动窗口限流，基于漏桶和令牌桶算法的限流。

从是否支持多机拓展上来讲，又分为单机限流和分布式限流。单机限流大多通过线程锁的方式实现，而分布式限流多借助于Redis等中间件。

简单计数器限流

通过维护单位时间内的请求次数来实现限流，当请求次数超过最大限制时拒绝访问。这种实现方式的好处是实现起来较为简单，缺点是可能会产生“毛刺”。如下图。

滑动窗口限流

滑动窗口也是维护单位时间内的请求次数，其与简单计数器的区别是，滑动窗口的粒度更细，将一个大的时间窗口划分为若干个小的时间窗口，通过滑动时间删除小的时间窗口，以此来避免简单计数器的“毛刺”问题。如下图。

基于漏桶算法限流

漏桶算法将流量放入一个固定容量的“漏斗”中，以恒定速度将流量进行输出。当“漏斗”装满时，拒绝掉涌入的流量。

基于令牌桶算法限流

令牌桶算法每隔一段时间就将一定量的令牌放入桶中，获取到令牌的请求直接访问后段的服务，没有获取到令牌的请求会被拒绝。同时令牌桶有一定的容量，当桶中的令牌数达到最大值后，不再放入令牌。

几种方案各有优劣，需要结合实际场景进行选型。

方案	优势	劣势
计数器	实现最为简单方便	"毛刺"现象
滑动窗口	应对突发流量能力强，可配置性强	取决于窗口粒度，非严格均匀，流量整形效果弱
漏桶	流量整形效果最好，输出流量最平滑（均匀输出）	应对突发流量效果差
令牌桶	相较漏桶，有一定的应对突发流量的能力	各方面都比较平庸，实现起来最为复杂
这里有一个流量整形的概念。所谓流量整形，是指流量经过我们的限流器后，其形状发生了变化，将短时间的大流量整形为长时间的平缓流量。而显然，通过计数器及滑动窗口的方式实现的限流，通过暴力拒绝掉部分流量，仅仅是对流量进行了“裁剪”，并没有对流量进行时间维度上的重新分配。而漏桶算法与令牌桶算法，通过一定的阻塞机制，真正改变了流量的时间分布，实现了一定的削峰填谷的效果。

限流器整体结构

整体设计思路上是通过注解实现对controller无侵入的限流，通过拦截请求，获取请求中相应的参数进行定制化的限流逻辑处理，并调用redis脚本进行是否限流的判断。

因此整体分为三部分代码

限流注解，主要是一些限流参数的指定
redis限流脚本，限流方法的具体实现
切面层，拦截请求，定制化限流逻辑，调用限流脚本实现限流。

限流注解如下：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {/*** 限流key* @return*/String key() default "rate:limiter";/*** 窗口允许最大请求数* @return*/long maxCount() default 10;/*** 窗口宽度，单位为ms* @return*/long winWidth() default 1000;/*** 限流提示语* @return*/String message() default "false";
}

这里的限流key只是一个基本key，对于特定的业务逻辑，可以有一些定制化的限流，如对于我的使用场景下，需要对不同的租户进行分开的限流，那么就可以在限流逻辑中对key进行一个定制化，以实现拓展的效果。下面是切面层。

@Component
@Aspect
@Slf4j
public class RateLimitAspect {@ResourceStringRedisTemplate stringRedisTemplate;private DefaultRedisScript<Long> getRedisScript;@PostConstructpublic void init() {getRedisScript = new DefaultRedisScript<>();getRedisScript.setResultType(Long.class);getRedisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("rateLimiterSlidingWindow.lua")));log.info("RateLimiter[分布式限流处理器]脚本加载完成");}@Pointcut("@annotation(com.tencent.cloud.iov.ivm.annotations.RateLimiter)")public void rateLimiter() {}@Around("@annotation(rateLimiter)")public Object around(ProceedingJoinPoint proceedingJoinPoint, RateLimiter rateLimiter) throws Throwable {if (log.isDebugEnabled()) {log.debug("RateLimiter[分布式限流处理器]开始执行限流操作");}Signature signature = proceedingJoinPoint.getSignature();if (!(signature instanceof MethodSignature)) {throw new IllegalArgumentException("the Annotation @RateLimiter must used on method!");}/*** 获取注解参数*//** 限流模块key*  按业务需求定制化处理*  这里用tenantId作为key的一部分，实现分租户限流的目的*/String limitKey = rateLimiter.key();RequestVo arg = (RequestVo)proceedingJoinPoint.getArgs()[0];limitKey += "-" + arg.getTenantId();Preconditions.checkNotNull(limitKey);/**时间窗口内可接受的最大请求次数*/Long maxCount = rateLimiter.maxCount();/**时间窗口宽度*/Long winWidth = rateLimiter.winWidth();if (log.isDebugEnabled()) {log.debug("RateLimiterHandler[分布式限流处理器]参数值为-maxCount={},winWidth={}", maxCount, winWidth);}// 限流提示语String message = rateLimiter.message();if (StringUtils.isBlank(message)) {message = "false";}/*** 执行Lua脚本*/List<String> keyList = new ArrayList();// 设置key值为注解中的值keyList.add(limitKey);/*** 调用脚本并执行*/log.info("keyList={}, maxCount={}, winWidth={}", keyList, maxCount, winWidth);Long result = stringRedisTemplate.execute(getRedisScript, keyList, maxCount.toString(), winWidth.toString());if (result == 0) {String msg = "由于超过窗口宽度=" + winWidth + "-允许" + limitKey + "的请求次数=" + maxCount + "[触发限流]";log.debug(msg);throw new BusinessException(BusinessCode.EXCEEDING_LIMIT_ERROR);}if (log.isDebugEnabled()) {log.debug("RateLimiterHandler[分布式限流处理器]限流执行结果-result={},请求[正常]响应", result);}return proceedingJoinPoint.proceed();}
}

限流脚本

这里将redis的限流方法解耦开，通过使用不同的脚本，以实现不同方案的限流。

简单计数器

首先是简单计数器的限流。
redis限流脚本如下：

--获取KEY
local key1 = KEYS[1]local val = redis.call('incr', key1)
local ttl = redis.call('ttl', key1)--获取ARGV内的参数并打印
local expire = ARGV[1]
local times = ARGV[2]redis.log(redis.LOG_DEBUG,tostring(times))
redis.log(redis.LOG_DEBUG,tostring(expire))redis.log(redis.LOG_NOTICE, "incr "..key1.." "..val);
if val == 1 thenredis.call('expire', key1, tonumber(expire))
elseif ttl == -1 thenredis.call('expire', key1, tonumber(expire))end
endif val > tonumber(times) thenreturn 0
endreturn 1

比较简单，主要是利用了Redis脚本的原子性，这里不再过多介绍了。

滑动窗口的实现方式

限流脚本如下：

redis.replicate_commands();--获取KEY
local key = KEYS[1]--获取ARGV内的参数并打印
local max_quantity = ARGV[1]
local window_width = ARGV[2]--获取当前时间及时间边界
local time = redis.call('TIME') --返回值为当前所过去的秒数，当前秒所过去的微秒数
local timestamp = time[1] * 1000 + math.floor(time[2] / 1000)local left_border = timestamp - window_width--移除窗口外的值
redis.call('zremrangebyscore', key, 0, left_border)--统计窗口内元素个数
local count = redis.call('zcard', key)if count < tonumber(max_quantity) thenredis.call('zadd', key, timestamp, timestamp)return 1
elsereturn 0
end

每次获取当前的时间戳，并移除时间窗口外的元素，随后判断当前是否出发限流，由于这里时间的粒度是毫秒，因此限流的效果还是比较平滑的。

这里要注意通过redis.replicate_commands()开启命令复制模式。这是因为redis在集群模式下，对于获取时间这种命令，由于到达各台机器的时间不一致，因此会出现数据不一致的问题。而采用命令复制模式，会直接复制时间值而非获取时间的命令。

漏桶实现方式

主要是利用了redis 4.0的cell命令。

--获取KEY
local key1 = KEYS[1]--获取ARGV内的参数并打印
local max_quantity = ARGV[1]
local window_width = ARGV[2]--这里漏桶的容量直接写死了，后续应该作为参数传入
local res = redis.call('CL.THROTTLE', key, 1000, max_quantity, window_width)if res[1] == 0 thenreturn 1
elsereturn 0
end

这里要注意redis-cell模块需要额外安装。