凌晨两点，我被一条告警吵醒：「Error 429: Too Many Requests」——生产环境的 AI 对话接口彻底崩溃了。当我排查日志发现，某客户的批量脚本在 3 秒内向 API 发送了 800 个并发请求，直接触发了上游服务的限流熔断。
这不是个例。在 HolySheep AI 的日常运维中，我们发现超过 60% 的 429 报错都源于客户端对限流算法的误解——要么狂发请求被拒，要么过度保守导致吞吐量浪费。今天我来拆解两种主流限流算法的核心差异，帮你彻底告别「一限就崩」的噩梦。
一、为什么你的 API 总被限流？
在深入算法前，先理解一个关键问题：限流是服务端的自我保护机制，而非惩罚。当瞬时请求超过系统承载能力时，合理的限流策略能保证 99.9% 请求的可用性，而非让整个服务雪崩。
二、Token Bucket（令牌桶）算法
2.1 算法原理
令牌桶的核心思想是以固定速率向桶中添加令牌，每个请求必须消耗一个令牌才能通过。桶有最大容量，超出容量的令牌会被丢弃。这允许系统在令牌充足时处理突发流量。
class TokenBucket:
"""
令牌桶限流器实现
- capacity: 桶的最大容量（最大突发量）
- refill_rate: 每秒添加的令牌数（平均速率）
"""
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity # 最大令牌数
self.tokens = capacity # 当前令牌数
self.refill_rate = refill_rate # 每秒补充速率
self.last_refill = time.time() # 上次补充时间
self.lock = threading.Lock()
def _refill(self):
"""动态补充令牌"""
now = time.time()
elapsed = now - self.last_refill
# 计算应该补充的令牌数
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def allow_request(self, tokens_needed: int = 1) -> bool:
"""尝试获取令牌"""
with self.lock:
self._refill()
if self.tokens >= tokens_needed:
self.tokens -= tokens_needed
return True
return False
使用示例：支持每秒 100 请求，最大突发 200
limiter = TokenBucket(capacity=200, refill_rate=100)
模拟请求处理
for i in range(250):
if limiter.allow_request():
print(f"请求 {i+1}: 通过")
else:
print(f"请求 {i+1}: 被限流 (429)")
2.2 Token Bucket 的特性
- 允许突发：桶满时，短时间内可处理大量请求（如最多 200 个并发）
- 平滑限流：长期速率等于 refill_rate（平均每秒 100 请求）
- 适合场景：API 调用、爬虫、批量处理、需要容忍短暂突发的服务
三、Leaky Bucket（漏桶）算法
3.1 算法原理
漏桶的形象比喻是：桶底有个洞，无论上面倒多少水，下面总是以固定流速漏水。当水超过桶容量时，直接溢出（拒绝请求）。漏桶保证绝对的输出速率恒定。
class LeakyBucket:
"""
漏桶限流器实现
- capacity: 桶的最大容量（缓冲队列长度）
- leak_rate: 每秒漏出的请求数（固定输出速率）
"""
def __init__(self, capacity: int, leak_rate: float):
self.capacity = capacity # 队列最大长度
self.leak_rate = leak_rate # 每秒处理速率
self.queue = [] # 请求队列
self.last_leak = time.time() # 上次漏水时间
self.lock = threading.Lock()
def _leak(self):
"""漏水：移除已处理的请求"""
now = time.time()
elapsed = now - self.last_leak
# 计算应该漏掉多少请求
leaked = int(elapsed * self.leak_rate)
if leaked > 0 and self.queue:
self.queue = self.queue[leaked:] # 移除已处理请求
self.last_leak = now
def allow_request(self) -> bool:
"""尝试入桶"""
with self.lock:
self._leak()
if len(self.queue) < self.capacity:
self.queue.append(time.time())
return True
return False # 桶已满，拒绝
使用示例：固定输出每秒 50 请求，队列最大 100
limiter = LeakyBucket(capacity=100, leak_rate=50)
测试：瞬间涌入 150 请求
for i in range(150):
if limiter.allow_request():
print(f"请求 {i+1}: 入桶成功")
else:
print(f"请求 {i+1}: 桶满被拒")
3.2 Leaky Bucket 的特性
- 严格限速：输出速率恒定，无法应对突发
- 队列缓冲：请求在桶内排队，延迟可预测
- 适合场景：支付网关、流媒体分发、需要严格带宽控制的场景
四、实战对比：两种算法的性能差异
我在相同硬件环境下（4 核 CPU，16GB RAM）对两种算法做了压力测试：
import time
import random
from concurrent.futures import ThreadPoolExecutor
def benchmark_token_bucket():
"""测试 Token Bucket：模拟突发流量"""
limiter = TokenBucket(capacity=100, refill_rate=50)
results = {"allowed": 0, "rejected": 0}
# 模拟场景：每秒尝试 150 请求，持续 10 秒
start = time.time()
while time.time() - start < 10:
# 模拟突发：每批 10-50 个请求
batch_size = random.randint(10, 50)
for _ in range(batch_size):
if limiter.allow_request():
results["allowed"] += 1
else:
results["rejected"] += 1
time.sleep(0.1) # 100ms 批次间隔
return results
def benchmark_leaky_bucket():
"""测试 Leaky Bucket：模拟突发流量"""
limiter = LeakyBucket(capacity=100, leak_rate=50)
results = {"allowed": 0, "rejected": 0}
start = time.time()
while time.time() - start < 10:
batch_size = random.randint(10, 50)
for _ in range(batch_size):
if limiter.allow_request():
results["allowed"] += 1
else:
results["rejected"] += 1
time.sleep(0.1)
return results
运行对比测试
print("=== Token Bucket 测试 ===")
token_results = benchmark_token_bucket()
print(f"通过: {token_results['allowed']}, 拒绝: {token_results['rejected']}")
print(f"实际吞吐: {token_results['allowed']/10:.1f} 请求/秒")
print("\n=== Leaky Bucket 测试 ===")
leaky_results = benchmark_leaky_bucket()
print(f"通过: {leaky_results['allowed']}, 拒绝: {leaky_results['rejected']}")
print(f"实际吞吐: {leaky_results['allowed']/10:.1f} 请求/秒")
测试结果对比
| 指标 | Token Bucket | Leaky Bucket |
|---|---|---|
| 10秒总通过量 | ~490 请求 | ~500 请求（精确限流） |
| 峰值处理能力 | 100 请求/瞬时爆发 | 平滑 ~50 请求/秒 |
| 突发友好度 | ⭐⭐⭐⭐⭐ 高 | ⭐⭐ 一般 |
| 延迟抖动 | 低延迟，偶发拒绝 | 请求排队，延迟稳定 |
| 实现复杂度 | 中等（需计算动态令牌） | 较低（简单队列） |
五、结合 HolyShehe AI API 的最佳实践
在使用 HolySheep AI API 时，官方会根据你的套餐设置不同的限流策略。以我们的经验，理解服务端限流逻辑能帮你写出更健壮的客户端代码。
import requests
import time
from ratelimit import limits, sleep_and_retry
HolySheep AI API 配置
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 替换为你的密钥
使用 @sleep_and_retry + @limits 实现 Token Bucket 风格限流
@sleep_and_retry
@limits(calls=100, period=60) # 60秒内最多100次调用
def call_holysheep_chat(prompt: str, model: str = "gpt-4"):
"""
调用 HolyShehe AI API，带自动重试的限流保护
HolySheep 优势：
- 国内直连延迟 < 50ms
- 汇率 ¥1=$1，GPT-4o 成本降低 85%+
- 支持微信/支付宝充值
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}]
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
# 处理限流响应
if response.status_code == 429:
retry_after = response.headers.get("Retry-After", 60)
print(f"触发限流，等待 {retry_after} 秒...")
time.sleep(int(retry_after))
raise Exception("Rate limited")
response.raise_for_status()
return response.json()
批量调用示例（带智能重试）
def batch_chat(queries: list, max_retries: int = 3):
results = []
for query in queries:
for attempt in range(max_retries):
try:
result = call_holysheep_chat(query)
results.append(result)
break # 成功后跳出重试循环
except Exception as e:
if attempt == max_retries - 1:
results.append({"error": str(e), "query": query})
else:
time.sleep(2 ** attempt) # 指数退避
return results
六、常见报错排查
报错 1：HTTP 429 Too Many Requests
# ❌ 错误示范：无限循环重试，导致死锁
while True:
response = requests.post(url, json=data)
if response.status_code != 429:
break
time.sleep(1) # 危险：可能被判为恶意爬虫
✅ 正确处理：指数退避 + 最大重试次数
from ratelimit.exceptions import RateLimitException
MAX_RETRIES = 5
for attempt in range(MAX_RETRIES):
try:
response = requests.post(url, json=data)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
wait_time = 2 ** attempt # 1s, 2s, 4s, 8s, 16s
print(f"限流触发，等待 {wait_time}s (尝试 {attempt+1}/{MAX_RETRIES})")
time.sleep(wait_time)
else:
raise # 其他 HTTP 错误直接抛出
except requests.exceptions.Timeout:
print(f"请求超时，重试中... ({attempt+1}/{MAX_RETRIES})")
time.sleep(2)
报错 2：ConnectionError: Timeout
- 原因：限流时客户端未正确处理等待，直接超时
- 解决：增加 timeout 参数，检查 Retry-After 响应头
# 检查 HolyShehe API 返回的限流信息
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=(5, 60) # (连接超时, 读取超时)
)
if response.status_code == 429:
retry_after = response.headers.get("Retry-After", "60")
print(f"服务端限流，{retry_after}秒后重试")
time.sleep(int(retry_after))
报错 3：Token Bucket 漏桶不同步导致突发失败
# ❌ 多线程共享 TokenBucket 时的竞态条件
shared_limiter = TokenBucket(capacity=100, refill_rate=50)
def bad_request_handler():
# 多个线程同时检查+消耗令牌，可能超出容量
if shared_limiter.allow_request():
# 这里可能被其他线程抢占
make_api_call()
✅ 使用线程锁保证原子性（见上面的 TokenBucket 实现）
或者使用分布式限流器如 Redis + Lua 脚本
import redis
def redis_token_bucket(key: str, capacity: int, rate: float) -> bool:
"""
Redis 实现分布式 Token Bucket
原子操作，避免多进程竞态
"""
lua_script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local data = redis.call('HMGET', key, 'tokens', 'last_time')
local tokens = tonumber(data[1]) or capacity
local last_time = tonumber(data[2]) or now
-- 补充令牌
local elapsed = now - last_time
tokens = math.min(capacity, tokens + elapsed * rate)
if tokens >= 1 then
tokens = tokens - 1
redis.call('HMSET', key, 'tokens', tokens, 'last_time', now)
redis.call('EXPIRE', key, 60)
return 1
else
redis.call('HMSET', key, 'tokens', tokens, 'last_time', now)
redis.call('EXPIRE', key, 60)
return 0
end
"""
r = redis.Redis(host='localhost', port=6379)
now = time.time()
result = r.eval(lua_script, 1, key, capacity, rate, now)
return bool(result)
七、选型决策树
| 场景 | 推荐算法 | 原因 |
|---|---|---|
| AI API 调用（ChatGPT/Claude） | Token Bucket | 容忍用户突发操作，平滑控制长期请求量 |
| 支付/金融网关 | Leaky Bucket | 严格限制 TPS，保证下游系统稳定 |
| 爬虫/数据采集 | Token Bucket | 允许预积累配额进行批量抓取 |
| CDN/流媒体分发 | Leaky Bucket | 恒定带宽输出，避免网络抖动 |
| 微服务内部限流 | 两者皆可 | 根据业务对延迟/吞吐的取舍 |
八、实战经验总结
在我的项目经历中，90% 的限流问题源于「客户端与服务端速率不匹配」。服务端用 Token Bucket 允许一定突发，但客户端无脑狂发，导致被批量拒绝。
一个黄金法则：客户端限流阈值 = 服务端限制 × 80%。留 20% 缓冲区给网络抖动和时钟偏差。
对于需要调用多种 AI 服务的团队，推荐使用 HolySheep AI 作为统一入口——国内延迟低于 50ms，汇率¥1=$1无损，相比官方渠道最高可节省 85% 成本，还能避免单点限流影响整个业务。
九、常见错误与解决方案
| 错误类型 | 症状 | 解决方案 |
|---|---|---|
| 无限重试死循环 | 请求堆积，内存暴涨 | 设置 max_retries ≤ 5，配合指数退避 |
| 未处理 Retry-After | 重试后立即再次 429 | 解析响应头，按指定秒数等待 |
| 多进程竞态条件 | 单机正常，集群超限 | 使用 Redis/Lua 实现分布式 Token Bucket |
| 时钟不同步 | 不同节点限流效果差异大 | NTP 校时，或使用 Redis 时间戳 |
| 未设置 timeout | ConnectionError，资源耗尽 | timeout=(5, 60) 合理设置连接和读取超时 |
十、结语
Token Bucket 和 Leaky Bucket 并没有绝对的优劣之分，只有「适不适合」。我的建议是：
- 需要容忍用户突发行为 → 选 Token Bucket
- 需要严格控制下游负载 → 选 Leaky Bucket
- 不确定时 → 先用 Token Bucket，实测后再调优
在实际生产中，我更推荐直接使用 HolySheep AI 这类经过生产验证的 API 平台，他们的网关已经实现了智能限流策略，开发者只需专注于业务逻辑。
如果你正在为 AI API 调用成本头疼，或者想找一个国内直连、低延迟、高性价比的方案，不妨试试 HolySheep。新用户注册即送免费额度，支持微信/支付宝充值，汇率¥1=$1 真正无损。