简介

前一篇文章：为了方便开发，我打算实现一个Redis 工具集 主要介绍了开发 Redis 工具集的 MQ(Stream数据结构做消息队列)、delay(延迟队列)功能，这篇文件主要分享一下使用 redis 如何做分布式限流的设计方案。

限流诉求

我希望有一个限流工具，它具备以下功能：

分布式限流，不是单机限流；尽量少的开发；使用灵活，可以对API（请求）限流，也可以对某一个方法限流；能够有多种维度限流，比如按照 IP限流、接口限流等；方便使用者拓展，也就是限流的维度可以由使用者确定；

我列举了一个限流工具应该具备的功能，我参考了 spring gateway 的限流方式，设计了 redis 限流工具集；

使用演示如何使用呢？

1.引入 Maven 依赖（目前可以下载代码上传到自己的私服或者本地仓库，后面会推到 Maven 中央仓库）

xml复制代码 <dependency>        <groupId>cn.org.wangchangjiu</groupId>        <artifactId>redis-util-spring-boot-starter</artifactId>        <version>1.0.0-SNAPSHOT</version>    </dependency>

2.配置文件（application.yaml）开启限流功能

yaml复制代码redis:  util:    limit:      enable: true

3.假如我想以 IP 限流，那么在 配置文件（application.yaml）中如下配置

yaml复制代码redis:  util:      limit:        configs:          - path: "/redis/*"     --- 拦截请求以/redis/*开头的这些url            replenishRate: 1     --- 每秒中增加的令牌数量            burstCapacity: 5     --- 桶容量            keyResolver: "ipKeyResolver"  --- 限流维度        enable: true

解释一下上面的配置：

configs ：是每一个限流的配置，你可以配置多个，也就是可以对不同的请求给以不同的限流维度；

path: 表示需要拦截匹配哪些 URL，假如说我需要对redis开头(/redis/*)的请求url，使用IP限流，那么对于所有以 redis开头的url都会被以IP维度限流了。 那如果我还配置了一个 /** 的限流配置，即如下配置： yaml复制代码 redis: util: limit: configs: - path: "/**" replenishRate: 1 burstCapacity: 10 keyResolver: "apiKeyResolver" --- 限流维度 - path: "/redis/*" --- 拦截请求以/redis/*开头的这些url replenishRate: 1 --- 每秒中增加的令牌数量 burstCapacity: 5 --- 桶容量 keyResolver: "ipKeyResolver" --- 限流维度 enable: true 对于这种情况，如果请求 url：/redis/limit，该被哪个限流方案生效呢？答案是，生效的是 ipKeyResolver， 因为 url：/redis/limit 最匹配path: "/redis/*";

以上就是所有限流的开发了，是不是很简单，也就说只要简单配置 yaml 文件不需要开发其他代码就可以实现以 ip 限流或者以 请求API限流（apiKeyResolver）。

以IP限流为例，演示效果

yaml中对于限流配置如下：

yaml复制代码redis:  util:      limit:        configs:          - path: "/redis/*"     --- 拦截请求以/redis/*开头的这些url            replenishRate: 1     --- 每秒中增加的令牌数量            burstCapacity: 5     --- 桶容量            keyResolver: "ipKeyResolver"  --- 限流维度        enable: true

请求接口实现：

jMeter 压测工具配置：

jMeter 压测结果：

分析一下： 我们配置的桶容量是5个，也就是说初始化会有5个令牌（最高也只有5个，每秒中生产1个，到达5个后会丢弃），由上图 jMeter 我配置的是20个线程，1秒内跑完，1秒内没跑完，没有产生新的令牌，所以只有初始化的5个令牌可以使用，因此只能有5个请求通过。

如何扩展限流维度

如果你只是想 以ip限流或者以接口限流，那使用内置的 ipKeyResolver 和 apiKeyResolver 就可以了，不需要额外的开发，经过上面的配置就行。

但是如果你有别的限流维度如何简单扩展呢？

案例：假如我有这样的一个需求：针对首页接口（/api/service/home），每个用户每秒限流10次，那该怎么做呢？

第一步，你需要开发一个keyResolver(即限流的维度)typescript复制代码@Bean public KeyResolver tokenKeyResolver(){ return request -> request.getHeader("token"); }然后，你需要把自己实现的 keyResolver 配置在 yamlyaml复制代码redis: util: limit: configs: - path: "/api/service/home" --- 拦截请求以/redis/*开头的这些url replenishRate: 1 --- 每秒中增加的令牌数量 burstCapacity: 10 --- 桶容量 keyResolver: "tokenKeyResolver" --- 限流维度 enable: true

就上面这样两步就可以扩展限流维度啦。

自定义 token（用户）维度限流演示

jMeter 压测工具配置：

jMeter 压测结果：

分析： burstCapacity = 10 ，初始化10个令牌，3秒钟产生 3 个令牌，共记 13个令牌。

如果我只想对方法限流如何使用？

案例1：我想对某个方法以 IP维度 限流，如何实现？

如下图，在需要被限流的方法上增加注解 @RedisRateLimitConfig 即可；

所以上面的案例意思是，令牌桶容量是10个，每秒中产生1个令牌，也就是说，每分钟可以调用这个方法 70 次；

案例2：我想对某个方法限流，但是限流的维度（限流key）是与方法参数相关的，如何实现？

如上图配置，keyResolver 支持配置某个Bean，也支持 配置SPEL。上面的配置是对于方法参数key为维度限流。

代码分析代码结构详细分析自动装配做了些啥？

自动装配 RedisLimitAutoConfiguration 主要是创建了一下Bean;

limitRedisTemplate: 类型为 RedisTemplate<String, Object> 的Bean，用于发送redis命令，注意修改序列化方式；redisRequestRateLimiterScript：类型为 RedisScript，用于加载 redis 脚本；RedisRateLimitAspect：aop 切面Bean,拦截被 @RedisRateLimitConfig 修饰的方法，用于基于方法的限流；RedisLimitHandlerInterceptor：请求拦截器，拦截配置的请求，以及对请求限流；ApiConfigResolver：请求配置解析器，把 yaml 配置文件里的配置解析成目标对象，用于对请求的限流；RedisRateLimiter：redis核心限流器，核心是将lua脚本发到redis中执行，判断是否能拿到请求令牌，其方法 isAllowed 是限流的结果；ipKeyResolver：内置基于IP限流策略；apiKeyResolver：内置基于api限流策略；基于请求配置限流

具体流程如下图：

ApiConfigResolver 在容器启动之后，会解析 yaml 配置文件，生成限流配置；

用户发起请求；请求被拦截器拦截，从请求配置解析器中获取配置；拦截器获取当前请求的限流配置；拦截器通过限流配置调用 redisRateLimiter 的 isAllowed 方法判断是否被限流；redisRateLimiter 通过redisTemplate向redis发送命令，执行lua脚本；redisRateLimiter 获取返回令牌，判断lua返回值是否为1（表示允许请求） 7.拦截器拿到redisRateLimiter返回的是否允许，允许则通过，否则返回 429；基于方法配置限流

基于方法和基于请求的类似，只是触发点不一样，核心切面代码：

核心lua脚本

这也是 springcloud gateway 限流的lua脚本。

lua复制代码--生产速率，每秒生产多少个令牌local rate = tonumber(ARGV[1])--令牌桶容量local capacity = tonumber(ARGV[2])--当前时间（秒级时间戳）local now = tonumber(ARGV[3])--每个请求消耗的令牌个数 固定为 1local requested = tonumber(ARGV[4])--填充时间=容量/生产速率local fill_time = capacity/rate--key过期时间设置为填充时间的2倍local ttl = math.floor(fill_time*2)-- 获取剩余令牌数量 , KEYS[1] redis key名，用于保存限流维度下剩余令牌数量，request_rate_limiter.{id}.tokenslocal last_tokens = tonumber(redis.call("get", KEYS[1]))--不存在key，则初始化令牌数量为最大容量,也就是说的 初始化时令牌数为桶容量if last_tokens == nil then  last_tokens = capacityend--最近获取令牌秒级时间戳local last_refreshed = tonumber(redis.call("get", KEYS[2]))if last_refreshed == nil then  last_refreshed = 0end--距离上次获取令牌时间相差多少秒local delta = math.max(0, now-last_refreshed)--计算当前令牌数量（考虑delta时间内生成的令牌个数=delta*速率），取 容量和 剩余令牌+生成令牌数 的最小值，也就是说，达到容量后，就令牌就丢弃了local filled_tokens = math.min(capacity, last_tokens+(delta*rate))--当前令牌数量是否大于1local allowed = filled_tokens >= requestedlocal new_tokens = filled_tokenslocal allowed_num = 0--允许访问，新令牌数量-1，allowed_num=1if allowed then  new_tokens = filled_tokens - requested  allowed_num = 1end--保存令牌个数和最近获取令牌时间redis.call("setex", KEYS[1], ttl, new_tokens)redis.call("setex", KEYS[2], ttl, now)return allowed_num

其他模块的设计细节后面再说，欢迎大家使用，也请大家多多提建议以及好的功能点，我都可以整合上去。

作者：程序员中的废物

链接：