令牌桶算法是网络流量整形和速率限制领域最经典的算法之一。它的基本思想是：系统以恒定速率向一个固定容量的桶中放入令牌，当请求到达时，必须从桶中取出一个令牌才能被处理；如果桶中没有令牌，则请求被拒绝或排队等待。桶的容量决定了系统允许的瞬时突发流量上限，而令牌的投放速率则决定了长期的平均处理速率。在分布式系统中，多个节点需要共享同一个令牌桶的状态，这就要求我们借助Redis等共享存储来实现全局一致的限流逻辑。
令牌桶算法的核心原理与流量整形机制
令牌桶算法的核心可以用两个参数来描述：容量（capacity）和补充速率（rate）。容量决定了桶最多能存储多少个令牌，也就是系统允许的最大瞬时突发流量；补充速率决定了每秒向桶中添加多少个令牌，也就是系统的长期平均处理速率。当请求到达时，如果桶中有足够的令牌，请求被放行并消耗相应数量的令牌；如果令牌不足，请求被拒绝。桶中未消耗的令牌会持续积累，直到达到容量上限，后续新增的令牌会被丢弃。
与漏桶算法相比，令牌桶算法最大的优势在于对突发流量的容忍能力。漏桶算法强制要求输出速率恒定，任何超出该速率的请求都会被直接丢弃；而令牌桶算法允许在桶中积累令牌，当突发流量到来时，如果桶中有足够的令牌储备，这些请求可以一次性被处理掉，不会造成不必要的拒绝。举个例子，假设一个API的限流配置为每秒100个请求，桶容量为200。在空闲期间，桶中会积累到200个令牌。当突然有150个请求同时到达时，令牌桶可以一次性放行全部150个请求，而漏桶算法只能放行100个，剩余50个被拒绝。这种特性使得令牌桶算法在API网关、云服务接口限流等场景中得到了广泛应用。
从数学模型的角度来看，在任意时刻t，桶中可用的令牌数T(t)可以表示为：T(t) = min(C, T(t_prev) + R * (t - t_prev))，其中C是容量，R是补充速率，t_prev是上一次请求的时间戳。这个公式表明，令牌的数量随时间线性增长，但不会超过桶的容量上限。每次请求消耗一个令牌后，T(t)减1。理解这个数学模型对于后续实现分布式限流器至关重要，因为我们需要在分布式环境中精确地计算这个值，并且保证多个节点同时操作时不会出现竞态条件。
分布式环境下的限流挑战与Redis方案
在单体应用中，实现令牌桶限流非常简单——只需要在内存中维护一个桶的状态即可。然而，当系统扩展为分布式集群后，问题变得复杂得多。假设集群中有三个节点，每个节点独立维护自己的令牌桶，那么全局的限流阈值实际上是单节点阈值的三倍，这显然不符合预期。如果将全局阈值均分到各节点，又会面临节点动态扩缩容时配额无法自适应的问题。更糟糕的是，当某个节点负载较高而其他节点空闲时，无法实现令牌的全局调度，导致资源利用率不均衡。
为了解决这些问题，必须引入一个共享存储来协调各节点的令牌分配。Redis是最常见的选择，原因有三：首先，Redis是单线程模型，天然保证了操作的原子性；其次，Redis提供了丰富的数据结构和Lua脚本支持，可以高效地实现令牌桶逻辑；最后，Redis的读写延迟通常在亚毫秒级别，对业务性能的影响可以接受。核心思路是将令牌桶的状态存储在Redis中，所有节点在处理请求前都先从Redis获取令牌，从而实现全局统一的限流控制。
然而，简单地用Redis的INCR命令来实现令牌桶是行不通的。令牌桶算法涉及读取当前令牌数、计算时间间隔内新增的令牌、判断是否足够消耗、更新令牌数等多个步骤，如果这些步骤不是原子性执行的，在并发场景下就会出现竞态条件，导致限流不准确。比如节点A和节点B同时读取到桶中有5个令牌，两者都判断令牌充足，各自消耗1个令牌后写回4，实际上应该写回3。解决方案是使用Redis的Lua脚本功能，将整个令牌桶逻辑封装在一个脚本中，由Redis保证原子执行。下面是一个完整的Lua脚本实现：
-- 令牌桶限流Lua脚本
-- KEYS[1]: 令牌桶的Redis键名
-- ARGV[1]: 桶容量
-- ARGV[2]: 令牌补充速率(令牌/秒)
-- ARGV[3]: 当前时间戳(秒,浮点数)
-- ARGV[4]: 请求消耗的令牌数
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])
-- 获取桶的当前状态
local bucket = redis.call('HMGET', key, 'tokens', 'timestamp')
local tokens = tonumber(bucket[1])
local last_time = tonumber(bucket[2])
-- 如果桶不存在,初始化为满桶
if tokens == nil then
tokens = capacity
last_time = now
end
-- 计算自上次请求以来新增的令牌
local delta = math.max(0, now - last_time)
local new_tokens = math.min(capacity, tokens + delta * rate)
-- 判断令牌是否足够
local allowed = new_tokens >= requested
if allowed then
new_tokens = new_tokens - requested
end
-- 更新桶状态
redis.call('HMSET', key, 'tokens', new_tokens, 'timestamp', now)
redis.call('EXPIRE', key, math.ceil(capacity / rate) * 2)
return {allowed and 1 or 0, new_tokens}Ruby实现完整的分布式令牌桶限流器
有了Lua脚本作为基础，接下来就可以用Ruby封装一个完整的分布式限流器。这个限流器需要具备以下能力：加载Lua脚本到Redis、在请求处理前调用脚本获取令牌、根据返回结果决定是否放行、以及提供配置管理和监控接口。我们将使用redis-rb这个Gem来与Redis交互，它对Lua脚本有良好的支持，能够通过EVALSHA命令高效地执行预加载的脚本。
下面是限流器的核心实现代码。我们定义一个TokenBucketLimiter类，它接受Redis连接、桶容量和补充速率作为初始化参数，提供一个limit方法用于判断请求是否被允许。该方法首先获取当前时间戳，然后通过EVALSHA命令调用预加载的Lua脚本，根据返回值做出决策。使用EVALSHA而非EVAL可以减少网络传输量，提升性能。同时，我们还实现了Redis不可用时的降级策略，避免Redis故障导致整个服务不可用。
require 'redis'
require 'time'
class TokenBucketLimiter
# 令牌桶Lua脚本
LUA_SCRIPT = <<~LUA
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])
local bucket = redis.call('HMGET', key, 'tokens', 'timestamp')
local tokens = tonumber(bucket[1])
local last_time = tonumber(bucket[2])
if tokens == nil then
tokens = capacity
last_time = now
end
local delta = math.max(0, now - last_time)
local new_tokens = math.min(capacity, tokens + delta * rate)
local allowed = new_tokens >= requested
if allowed then
new_tokens = new_tokens - requested
end
redis.call('HMSET', key, 'tokens', new_tokens, 'timestamp', now)
redis.call('EXPIRE', key, math.ceil(capacity / rate) * 2)
return {allowed and 1 or 0, new_tokens}
LUA
def initialize(redis_url:, capacity:, rate:)
@redis = Redis.new(url: redis_url)
@capacity = capacity
@rate = rate
# 预加载Lua脚本,获取SHA1哈希
@script_sha = @redis.script(:load, LUA_SCRIPT)
end
# 判断请求是否被允许通过
# key: 限流标识,如用户ID或API路径
# tokens: 本次请求消耗的令牌数,默认为1
# 返回 [allowed, remaining_tokens]
def limit(key, tokens = 1)
now = Time.now.to_f
bucket_key = "token_bucket:#{key}"
result = @redis.evalsha(
@script_sha,
keys: [bucket_key],
argv: [@capacity, @rate, now, tokens]
)
allowed = result[0] == 1
remaining = result[1]
[allowed, remaining]
rescue Redis::BaseError => e
# Redis不可用时的降级策略:直接放行
# 避免Redis故障导致服务不可用
warn "Redis error in rate limiter: #{e.message}"
[true, @capacity]
end
end使用这个限流器非常简单。假设我们要对某个API接口实施限流，每秒最多允许100个请求，允许的突发流量上限为200。首先创建一个限流器实例，然后在每个请求处理前调用limit方法即可。limit方法返回一个数组，第一个元素是布尔值表示是否放行，第二个元素是当前桶中剩余的令牌数，可以用于在响应头中返回给客户端，帮助客户端了解限流状态并做出相应的退避策略。下面是一个在Rack中间件中使用该限流器的示例：
# 初始化限流器
limiter = TokenBucketLimiter.new(
redis_url: 'redis://127.0.0.1:6379',
capacity: 200,
rate: 100 # 每秒补充100个令牌
)
# 在Rack中间件中使用
class RateLimitMiddleware
def initialize(app, limiter)
@app = app
@limiter = limiter
end
def call(env)
# 以客户端IP作为限流标识
client_ip = env['REMOTE_ADDR']
allowed, remaining = @limiter.limit(client_ip)
unless allowed
return [429, {'Content-Type' => 'application/json'}, ['{"error":"rate limit exceeded"}']]
end
status, headers, body = @app.call(env)
# 在响应头中返回限流信息
headers['X-RateLimit-Remaining'] = remaining.to_s
[status, headers, body]
end
end生产环境中的优化策略与最佳实践
上述实现虽然功能完整，但在生产环境中还需要考虑多个方面的优化。首先是Redis的高可用问题。如果Redis发生故障，限流器将无法正常工作。代码中的降级策略是直接放行所有请求，这在某些场景下可能不可接受——比如对外收费的API，放行超额请求意味着经济损失。替代方案是在本地维护一个备用的令牌桶，当Redis不可用时切换到本地限流模式。虽然本地限流无法保证全局准确性，但至少能提供基本的保护。可以使用一个简单的内存级令牌桶作为降级方案，其容量和速率可以设置为全局配额除以预估节点数。
其次是性能优化。每次请求都访问Redis会引入额外的网络延迟，在高并发场景下可能成为瓶颈。一种优化策略是批量获取令牌：每个节点定期从Redis一次性获取一批令牌（比如10个），缓存在本地内存中，本地请求优先消耗缓存的令牌，只有当本地令牌耗尽时才再次访问Redis。这种方案将Redis的访问频率降低了N倍（N为批量大小），但代价是全局限流的精度会有所下降，因为各节点缓存的令牌实际上是预借的。需要在性能和精度之间做出权衡。对于大多数业务场景来说，批量大小设置为5到10是一个比较合理的起点。
第三是动态配置的需求。在生产环境中，限流参数往往需要根据系统负载动态调整。比如在促销活动期间需要临时提高限流阈值，活动结束后再恢复。为此，可以将限流参数存储在Redis或配置中心中，限流器定期拉取最新配置并更新自身的容量和速率参数。下面是一个支持动态配置的限流器扩展实现，它在每次限流检查时顺带检查配置是否更新，确保参数调整能够实时生效：
class DynamicTokenBucketLimiter < TokenBucketLimiter
def initialize(redis_url:, config_key:, default_capacity:, default_rate:)
super(redis_url: redis_url, capacity: default_capacity, rate: default_rate)
@config_key = config_key
@default_capacity = default_capacity
@default_rate = default_rate
@last_config_check = 0
@config_check_interval = 10 # 每10秒检查一次配置
end
def limit(key, tokens = 1)
check_config_update
super
end
private
def check_config_update
now = Time.now.to_i
return if now - @last_config_check < @config_check_interval
@last_config_check = now
config = @redis.hgetall(@config_key)
if config['capacity'] && config['rate']
@capacity = config['capacity'].to_i
@rate = config['rate'].to_f
else
@capacity = @default_capacity
@rate = @default_rate
end
end
end最后需要强调的是监控和告警。限流器作为系统的关键组件，必须对其运行状态进行持续监控。关键指标包括：限流触发率（被拒绝的请求占总请求的比例）、令牌桶利用率（剩余令牌占容量的比例）、Redis操作延迟、降级触发次数等。这些指标可以通过Prometheus等监控系统采集，并设置合理的告警阈值。当限流触发率突然飙升时，可能意味着上游流量异常或限流配置不合理，需要及时介入处理。当降级频繁触发时，说明Redis的稳定性存在问题，需要排查Redis集群的健康状态。通过完善的监控体系，才能确保限流器在生产环境中稳定可靠地运行，真正发挥流量整形和保护后端服务的作用。