基于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";
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

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

@Component
@Aspect
@Slf4j
public class RateLimitAspect {

    @Resource
    StringRedisTemplate stringRedisTemplate;

    private DefaultRedisScript getRedisScript;

    @PostConstruct
    public 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 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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

限流脚本

这里将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 then
    redis.call('expire', key1, tonumber(expire))
else
    if ttl == -1 then
        redis.call('expire', key1, tonumber(expire))
    end
end

if val > tonumber(times) then
    return 0
end

return 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

比较简单，主要是利用了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) then
    redis.call('zadd', key, timestamp, timestamp)
    return 1
else
    return 0
end
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

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

这里要注意通过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 then
    return 1
else
    return 0
end
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

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