soul源码解读（十六）
- 使用rateLimiter
- 分析源码
官网的介绍和流程图如下：
- 限流插件，是网关对流量管控限制核心的实现。
- 可以到接口级别，也可以到参数级别，具体怎么用，还得看你对流量配置。
使用 rateLimiter 插件，我们需要安装 redis ，这里可以参考我之前写的文章Windows下部署redis主从、哨兵（sentinel）、集群（cluster）
使用rateLimiter
1.启动 admin，打开 rateLimiter 插件开关
2.在 bootstrap 项目的 pom 文件引入 rateLimiter 插件的相关依赖，启动 bootstrap
<dependency>
<groupId>org.dromara</groupId>
<artifactId>soul-spring-boot-starter-plugin-ratelimiter</artifactId>
<version>${project.version}</version>
</dependency>
3.启动一个 http 服务，这里我们启动 soul-examples-http 服务
4.配置限流参数
在 admin 后台选择 rateLimiter 插件，添加选择器
添加规则
- 容量：是允许用户在一秒钟内执行的最大请求数。这是令牌桶可以保存的令牌数。
- 速率：是你允许用户每秒执行多少请求，而丢弃任何请求。这是令牌桶的填充速率。
5.测试接口
在 postman 新建测试用例，编写一个响应码=200的断言
点击测试用例右侧小箭头，点击 run
设置并发数为100，延迟为0
点击运行，可以看到在执行成10次请求后，接口返回异常了。
分析源码
接下来我们分下下 rateLimiter 插件是怎么实现限流的。
刚刚用 postman 测试接口之后，我们可以看到控制台有输出下面的日志
rate_limiter selector success match , selector name :http限流
rate_limiter rule success match , rule name :限流findById
RateLimiter response:Response{allowed=true, tokensRemaining=9}
我们找到日志输出的地方，发现是 AbstractSoulPlugin#execute。
根据前面分析过的源码，我们知道 soul 会遍历所有插件，如果插件不用跳过，就会执行这个函数。
这个函数又会判断插件有没有开启，有没有匹配的选择器和规则。
我们进到 RateLimiterPlugin.java 里去看下这个插件是怎么工作的。
// RateLimiterPlugin.java
protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
...
// 这里会判断是否允许请求通过，允许就执行下一个插件逻辑，不允许就返回 Too Many Requests
return redisRateLimiter.isAllowed(rule.getId(), limiterHandle.getReplenishRate(), limiterHandle.getBurstCapacity())
.flatMap(response -> {
if (!response.isAllowed()) {
exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
Object error = SoulResultWrap.error(SoulResultEnum.TOO_MANY_REQUESTS.getCode(), SoulResultEnum.TOO_MANY_REQUESTS.getMsg(), null);
return WebFluxResultUtils.result(exchange, error);
}
return chain.execute(exchange);
});
}
我们继续看下 isAllowed 函数
// RedisRateLimiter.java
public Mono<RateLimiterResponse> isAllowed(final String id, final double replenishRate, final double burstCapacity) {
...
List<String> keys = getKeys(id);
List<String> scriptArgs = Arrays.asList(replenishRate + "", burstCapacity + "", Instant.now().getEpochSecond() + "", "1");
Flux<List<Long>> resultFlux = Singleton.INST.get(ReactiveRedisTemplate.class).execute(this.script, keys, scriptArgs);
return resultFlux.onErrorResume(throwable -> Flux.just(Arrays.asList(1L, -1L)))
.reduce(new ArrayList<Long>(), (longs, l) -> {
longs.addAll(l);
return longs;
}).map(results -> {
// 判断请求是否允许通过
boolean allowed = results.get(0) == 1L;
Long tokensLeft = results.get(1);
RateLimiterResponse rateLimiterResponse = new RateLimiterResponse(allowed, tokensLeft);
log.info("RateLimiter response:{}", rateLimiterResponse.toString());
return rateLimiterResponse;
}).doOnError(throwable -> log.error("Error determining if user allowed from redis:{}", throwable.getMessage()));
}
// 组装keys
private static List<String> getKeys(final String id) {
String prefix = "request_rate_limiter.{" + id;
String tokenKey = prefix + "}.tokens";
String timestampKey = prefix + "}.timestamp";
return Arrays.asList(tokenKey, timestampKey);
}
// 获取lua脚本
private RedisScript<List<Long>> redisScript() {
...
redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("/META-INF/scripts/request_rate_limiter.lua")));
return redisScript;
}
可以看到，上面是用 lua 脚本来保证操作的原子性的。
具体 lua 脚本在 soul-plugin-ratelimiter 模块 /src/main/resource/META-INF/scripts/request_rate_limiter.lua
lua 脚本最后返回一个 Long 集合，第一个数用来标识是否允许请求通过 1 通过 0 不通过，第二个数表示剩余的容量。
我们执行一次请求，可以看到 redis 里新建了两个 key
127.0.0.1:6379> keys *
1) "request_rate_limiter.{1356226169225809920}.tokens"
2) "request_rate_limiter.{1356226169225809920}.timestamp"
rateLimiter 原理总结：
请求过来之后，会通过 lua 脚本在 redis 创建两个 key，通过一次请求就减少1个容量，同时会按照设定的
速率补充容量，去过请求太快，就返回不允许请求通过。