简介
前一篇文章：为了方便开发，我打算实现一个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 配置在 yaml
- yaml复制代码
- 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]) --每个请求消耗的令牌个数 固定为 1 local 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}.tokens local last_tokens = tonumber(redis.call("get", KEYS[1])) --不存在key，则初始化令牌数量为最大容量,也就是说的 初始化时令牌数为桶容量 if last_tokens == nil then last_tokens = capacity end --最近获取令牌秒级时间戳 local last_refreshed = tonumber(redis.call("get", KEYS[2])) if last_refreshed == nil then last_refreshed = 0 end --距离上次获取令牌时间相差多少秒 local delta = math.max(0, now-last_refreshed) --计算当前令牌数量（考虑delta时间内生成的令牌个数=delta*速率），取 容量和 剩余令牌+生成令牌数 的最小值，也就是说，达到容量后，就令牌就丢弃了 local filled_tokens = math.min(capacity, last_tokens+(delta*rate)) --当前令牌数量是否大于1 local allowed = filled_tokens >= requested local new_tokens = filled_tokens local allowed_num = 0 --允许访问，新令牌数量-1，allowed_num=1 if allowed then new_tokens = filled_tokens - requested allowed_num = 1 end --保存令牌个数和最近获取令牌时间 redis.call("setex", KEYS[1], ttl, new_tokens) redis.call("setex", KEYS[2], ttl, now) return allowed_num
后记
其他模块的设计细节后面再说，欢迎大家使用，也请大家多多提建议以及好的功能点，我都可以整合上去。
作者：程序员中的废物
