深入探索令牌桶限流的原理与实践
在当今的互联网时代，随着用户数量和请求量的不断增加，系统的性能和稳定性面临着巨大的挑战。限流算法作为保障系统稳定性的重要手段之一，被广泛应用于各种服务和应用中。限流的核心目的是对某一时间窗口内的请求数进行限制，保持系统的可用性和稳定性，防止因流量暴增而导致的系统运行缓慢或宕机。
常见限流算法对比
常见的限流算法有四种：
● 令牌桶算法（Token Bucket）
· 原理：令牌桶算法是一种基于令牌的限流算法，它维护一个固定容量的令牌桶，按照固定速率往桶中添加令牌，每当有请求到来时，消耗一个令牌，如果桶中没有足够的令牌，则拒绝该请求。
· 优点：平滑限流，可以应对突发流量；灵活控制流量速率。
· 缺点：对于突发流量，需要足够的令牌桶容量用来应对，否则可能会出现丢弃部分请求的情况。
● 漏桶算法（Leaky Bucket）
· 原理：漏桶算法是一种基于漏桶的限流算法，它维护一个固定容量的漏桶，按照固定速率漏水，当有请求到来时，漏桶中的水量增加，如果漏桶已满，则拒绝该请求。
· 优点：平滑限流，可以限制请求的处理速率。
· 缺点：对于突发流量，可能会出现短时的请求丢失。
● 计数器算法（Fixed Window）
· 原理：计数器算法是一种简单的限流算法，通过统计单位时间内的请求数量，当请求达到设定的阈值时，拒绝后续请求。
· 优点：简单易实现，对于固定流量场景适用。
· 缺点：无法处理突发流量，可能导致性能下降或服务不可用。
● 漏斗算法（Funnel Algorithm）
· 原理：漏斗算法是一种基于漏斗的限流算法，它模拟了一个漏斗的过程，请求进入漏斗后，根据漏斗的容量和速率来控制请求的处理速度。
· 优点：可以动态调整漏斗的容量和速率，适应不同的流量情况。
· 缺点：相对复杂，实现和调优难度较大。
令牌桶算法
在微服务架构中，Spring Cloud Gateway 作为服务网关，承担着流量管理和路由分发的关键角色。为了增强其限流能力，Spring Cloud Gateway 官方提供了 RequestRateLimiterGatewayFilterFactory 类，该类通过集成 Redis 和使用 Lua 脚本来实现高效的令牌桶限流策略。
在众多限流算法中，令牌桶算法以其独特的优点和灵活性，成为了业界广泛采用的一种方案。数栈「数据服务平台DataAPI」基于 Spring Cloud 的已有限流能力，进一步进行了定制化的开发和优化。
上文简单介绍过，令牌桶算法是一种流量控制算法，用于限制单位时间内通过的请求数量，它能够平滑地限制请求速率，防止突发大量请求对系统造成过载。以下是令牌桶算法的基本原理和工作流程：
● 令桶牌
算法中的“令牌桶”类似于一个存放令牌的容器。在每个固定的时间间隔（例如每秒），系统会往令牌桶中添加一定数量的令牌。
● 请求处理
当一个请求到达时，系统会尝试从令牌桶中获取一个令牌。如果令牌桶中有足够的令牌，则系统允许处理该请求，并从令牌桶中取走一个令牌。如果令牌桶中没有足够的令牌，则系统拒绝该请求或者将其放入队列等待。
● 令牌生成
令牌桶中的令牌数量会受到限制，它不能无限制地增长。当令牌桶已经满了时，不会再继续产生新的令牌，多余的令牌会被丢弃。
通过上述工作流程，令牌桶算法可以实现对请求的流量控制，使得系统能够以稳定的速率处理请求。这种算法适合于需要限制请求速率的场景，如接口调用频率控制、网络流量控制等。
自定义限流
通过继承 AbstractRateLimiter 重写 isAllowed 方法实现自定义限流。
通过继承 AbstractGatewayFilterFactory 重写 apply 方法调用具体的限流方法。
isAllowed 方法分析：
@Override
@SuppressWarnings("unchecked")
public Mono<Response> isAllowed(String routeId, String id) {
// 初始化判断
if (!this.initialized.get()) {
throw new IllegalStateException("RedisRateLimiter is not initialized");
}
// How many requests per second do you want a user to be allowed to do?
int replenishRate = 0;
// How much bursting do you want to allow?
int burstCapacity = 0;
// How many tokens are requested per request?
int requestedTokens = 1;
final RedisLimitDefinition limitDefinition;
if(LimitListRouteDefinitionLocator.ID_PREFIX.equals(routeId.split(ValidateRequestGlobalFilter.SPLIT)[0])){
limitDefinition = limitConfig.getLimitInfoMap().get(id);
if (ObjectUtils.isEmpty(limitDefinition) ){
throw new IllegalArgumentException("No Configuration found for uri " + id);
}
replenishRate = limitDefinition.getReplenishRate();
// How much bursting do you want to allow?
burstCapacity = limitDefinition.getBurstCapacity();
} else{
// 自定义逻辑，分为分钟级限流和秒级限流
String[] routeIdArr = routeId.split(ValidateRequestGlobalFilter.SEPARATOR);
if(routeIdArr.length > 1){
replenishRate = Integer.parseInt(routeIdArr[0]);
burstCapacity = replenishRate * 60;
limitDefinition = new RedisLimitDefinition(replenishRate, replenishRate, null, null);
requestedTokens = 60;
} else{
replenishRate = Integer.parseInt(routeIdArr[0]);
burstCapacity = replenishRate;
limitDefinition = new RedisLimitDefinition(replenishRate, replenishRate, null, null);
requestedTokens = 1;
}
}
try {
List<String> keys = getKeys(id);
// The arguments to the LUA script. time() returns unixtime in seconds.
List<String> scriptArgs = Arrays.asList(replenishRate + "",
burstCapacity + "", Instant.now().getEpochSecond() + "",
requestedTokens + "");
// allowed, tokens_left = redis.eval(SCRIPT, keys, args)
// 调用lua脚本并返回结果
Flux<List<Long>> flux = this.reactiveStringRedisTemplate.execute(this.script, keys,
scriptArgs);
return flux.onErrorResume(throwable -> {
if (log.isDebugEnabled()) {
log.debug("Error calling rate limiter lua", throwable);
}
return 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);
Response response = new Response(allowed,
getHeaders(limitDefinition, tokensLeft));
if (log.isDebugEnabled()) {
log.debug("response: " + response);
}
return response;
});
}
catch (Exception e) {
/*
* We don't want a hard dependency on Redis to allow traffic. Make sure to set
* an alert so you know if this is happening too much. Stripe's observed
* failure rate is 0.01%.
*/
log.error("Error determining if user allowed from redis", e);
}
return Mono.just(new Response(true, getHeaders(limitDefinition, -1L)));
}
LUA 脚本
LUA 脚本是内置在数栈后端服务中的，位置如图所示：
接下来进行脚本代码分析：
-- 基于唯一id生成的token key和timestamp key
local tokens_key = KEYS[1]
local timestamp_key = KEYS[2]
--redis.log(redis.LOG_WARNING, "tokens_key " .. tokens_key)
-- 对应的脚本入参，分别为
-- 往令牌桶放置的速率，每秒n个
local rate = tonumber(ARGV[1])
-- 令牌桶最大容量
local capacity = tonumber(ARGV[2])
-- 当前时间戳，秒级
local now = tonumber(ARGV[3])
-- 每次请求消耗的令牌数量
local requested = tonumber(ARGV[4])
-- 放满令牌桶是的时长
local fill_time = capacity/rate
-- redis key过期时间
local ttl = math.floor(fill_time*2)
--redis.log(redis.LOG_WARNING, "rate " .. ARGV[1])
--redis.log(redis.LOG_WARNING, "capacity " .. ARGV[2])
--redis.log(redis.LOG_WARNING, "now " .. ARGV[3])
--redis.log(redis.LOG_WARNING, "requested " .. ARGV[4])
--redis.log(redis.LOG_WARNING, "filltime " .. fill_time)
--redis.log(redis.LOG_WARNING, "ttl " .. ttl)
-- 这里对于last_tokens和last_refreshed的操作既是赋值也有初始化的作用
-- 获取令牌桶的数量，如果为空，将令牌桶容量赋值
local last_tokens = tonumber(redis.call("get", tokens_key))
if last_tokens == nil then
last_tokens = capacity
end
--redis.log(redis.LOG_WARNING, "last_tokens " .. last_tokens)
-- 获取最后的更新时间戳，如果为空，设置为0
local last_refreshed = tonumber(redis.call("get", timestamp_key))
if last_refreshed == nil then
last_refreshed = 0
end
--redis.log(redis.LOG_WARNING, "last_refreshed " .. last_refreshed)
-- 计算出上次到这次统计的时间间隔
local delta = math.max(0, now-last_refreshed)
-- 上次剩余的令牌数量+新生成的令牌，这个值不会大于令牌桶容量
local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
-- 总剩余令牌是否能满足一次请求消耗的令牌
local allowed = filled_tokens >= requested
local new_tokens = filled_tokens
-- 请求是否允许通过的标志位
local allowed_num = 0
-- 满足的话将总数减去单次消耗的令牌记为当前桶容量，并重新放置到redis key中
if allowed then
new_tokens = filled_tokens - requested
allowed_num = 1
end
--redis.log(redis.LOG_WARNING, "delta " .. delta)
--redis.log(redis.LOG_WARNING, "filled_tokens " .. filled_tokens)
--redis.log(redis.LOG_WARNING, "allowed_num " .. allowed_num)
--redis.log(redis.LOG_WARNING, "new_tokens " .. new_tokens)
if ttl > 0 then
redis.call("setex", tokens_key, ttl, new_tokens)
redis.call("setex", timestamp_key, ttl, now)
end
-- return { allowed_num, new_tokens, capacity, filled_tokens, requested, new_tokens }
return { allowed_num, new_tokens }
模拟计算流程：
针对上述流程，由于设定的是每分钟允许三次请求，可以看到当桶内令牌满是只能通过三次请求。当桶内令牌空时，每隔20秒会生成刚好够一次请求的令牌数，实现对流量的平滑控制。
《行业指标体系白皮书》下载地址：https://www.dtstack.com/resources/1057?src=szsm
《数栈产品白皮书》下载地址：https://www.dtstack.com/resources/1004?src=szsm
《数据治理行业实践白皮书》下载地址：https://www.dtstack.com/resources/1001?src=szsm
想了解或咨询更多有关大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=szbky