当前位置：AIGC资讯 > AIGC > 正文

深入探索令牌桶限流的原理与实践

在当今的互联网时代，随着用户数量和请求量的不断增加，系统的性能和稳定性面临着巨大的挑战。限流算法作为保障系统稳定性的重要手段之一，被广泛应用于各种服务和应用中。限流的核心目的是对某一时间窗口内的请求数进行限制，保持系统的可用性和稳定性，防止因流量暴增而导致的系统运行缓慢或宕机。

常见限流算法对比

常见的限流算法有四种：

● 令牌桶算法（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=szkyzg

tokendefiscripteltdebug白皮书自定义稳定性parseappfix时间戳asl工作流工作流程cto计数器广泛应用etl关键角色

更新时间 2024-05-30