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

常见限流算法对比

常见的限流算法有四种：

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