做量化交易或加密货币数据采集的开发者,十有八九会遇到一个令人头疼的问题:API Rate Limit(速率限制)。当你满怀信心地跑起策略,却发现交易所返回 429 错误代码,订单发不出去,数据采集中断,那种感觉就像在高速公路上被交警拦下一样——明明什么都没做错,却被告知"你太快了"。
我在早期开发加密货币交易机器人时,也被这个问题折磨了整整两周。当时完全不懂什么是 rate limit,只会傻傻地重试,结果账号被临时封禁,错过了绝佳的套利窗口。这篇文章,就是我用无数次报错换来的实战经验总结,帮助你从零掌握 rate limit 的应对之道。
什么是 Rate Limit?为什么交易所要限速?
简单来说,Rate Limit 就是交易所对你 API 请求频率的限制。每个交易所都有自己的规则,比如 Binance 允许每秒最多 1200 个请求,OKX 的公开接口是每秒 20 次,Bybit 则根据接口类型分为不同档位。
交易所设置 rate limit 主要是为了三个目的:
- 保护服务器稳定:防止恶意爬虫或失控程序把服务器打垮
- 保证公平交易:避免高频交易者利用速度优势碾压普通用户
- 控制资源成本:AWS 服务器不便宜,交易所也要精打细算
理解这一点很重要:rate limit 不是针对你,而是保护整个交易生态的必需机制。把它想象成地铁高峰期的限流闸机——不是为难你,是为了让大家都能有序乘车。
主流交易所 Rate Limit 规则一览
| 交易所 | 权重接口限制 | 订单频率限制 | 封禁策略 |
|---|---|---|---|
| Binance | 1200 次/分钟 | 10 次/秒 | 阶梯式封禁(1分钟~24小时) |
| OKX | 20 次/秒(公开) | 100 次/2秒 | IP + API Key 双重封禁 |
| Bybit | 6000 次/5分钟 | 50 次/秒 | 10分钟冷却期 |
| Deribit | 60 次/分钟 | 10 次/秒 | 递增式封禁 |
以上数据基于 2025 年第一季度各交易所官方文档整理,实际限制可能因账户等级不同而有所差异。建议开发者在正式使用前,务必查阅对应交易所的最新 API 文档。
三大经典 Rate Limit 处理策略
策略一:令牌桶算法(Token Bucket)
令牌桶是最常用的限流算法,想象一个漏水的桶:桶以固定速率漏水,同时以固定速率补充令牌。你的请求必须拿到令牌才能发出,没有令牌就只能等待。
这种策略的优点是允许一定程度的突发流量——你可以一次性消耗多个令牌,而不是严格匀速。它的缺点是实现相对复杂,需要维护一个桶状态。
import time
import threading
from collections import deque
class TokenBucket:
"""令牌桶限流器"""
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 acquire(self, tokens: int = 1, timeout: float = 30) -> bool:
"""尝试获取令牌,超时返回False"""
start_time = time.time()
while True:
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
# 计算需要等待多久才能补充足够令牌
needed = tokens - self.tokens
wait_time = needed / self.refill_rate
if time.time() - start_time + wait_time > timeout:
return False
time.sleep(0.01) # 避免CPU空转
def _refill(self):
"""补充令牌"""
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity,
self.tokens + elapsed * self.refill_rate)
self.last_refill = now
使用示例:创建每分钟1200次请求的限流器
rate_limiter = TokenBucket(capacity=1200, refill_rate=20) # 每秒补充20个
获取一个令牌后再发送请求
if rate_limiter.acquire():
response = requests.get("https://api.binance.com/api/v3/order",
headers={"X-MBX-APIKEY": API_KEY})
else:
print("请求超时:无法在规定时间内获取令牌")
策略二:滑动窗口算法(Sliding Window)
滑动窗口算法像是一个不断滑动的观察窗口,它记录最近 N 秒内的所有请求。当你准备发起新请求时,先检查这个窗口内已经有多少请求,如果已经满了就等待。
这种策略的优势是精度高,能够真正做到"最近 N 秒内不超过 X 次"的效果。缺点是需要占用额外内存来存储请求时间戳。
import time
from collections import deque
import threading
class SlidingWindowRateLimiter:
"""滑动窗口限流器"""
def __init__(self, max_requests: int, window_seconds: float):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque()
self.lock = threading.Lock()
def is_allowed(self) -> bool:
"""检查是否可以发起请求(不等待)"""
with self.lock:
now = time.time()
cutoff = now - self.window_seconds
# 清理过期的请求记录
while self.requests and self.requests[0] < cutoff:
self.requests.popleft()
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
return False
def acquire(self) -> float:
"""获取许可,返回需要等待的秒数"""
with self.lock:
now = time.time()
cutoff = now - self.window_seconds
# 清理过期记录
while self.requests and self.requests[0] < cutoff:
self.requests.popleft()
if len(self.requests) < self.max_requests:
self.requests.append(now)
return 0.0
# 返回需要等待的时间
oldest = self.requests[0]
wait_time = oldest + self.window_seconds - now
return max(0, wait_time)
使用示例:Binance 1200次/分钟的限制
limiter = SlidingWindowRateLimiter(max_requests=1200, window_seconds=60)
def safe_request(url: str, params: dict):
"""安全的API请求封装"""
wait_time = limiter.acquire()
if wait_time > 0:
print(f"限流等待 {wait_time:.2f} 秒...")
time.sleep(wait_time)
return requests.get(url, params=params)
策略三:指数退避重试(Exponential Backoff)
当请求被限流拒绝后,指数退避是最常用的恢复策略。核心思想很简单:第一次失败后等 1 秒重试,还失败就等 2 秒,再失败等 4 秒……以此类推,直到达到最大等待时间。
这种策略的好处是实现简单、对服务器友好。缺点是可能会错失短暂的机会窗口,不适合对延迟敏感的高频策略。
import time
import random
import requests
def exponential_backoff_request(
url: str,
headers: dict = None,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0
) -> dict:
"""
带指数退避的重试机制
参数:
url: 请求URL
headers: 请求头
max_retries: 最大重试次数
base_delay: 基础延迟秒数
max_delay: 最大延迟秒数
"""
for attempt in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=10)
if response.status_code == 200:
return {"success": True, "data": response.json()}
# 处理不同的HTTP错误码
if response.status_code == 429:
# Rate Limit 触发
retry_after = response.headers.get('Retry-After')
if retry_after:
wait_time = int(retry_after)
else:
# 指数退避计算
wait_time = min(base_delay * (2 ** attempt), max_delay)
# 添加随机抖动,避免多客户端同时重试
wait_time += random.uniform(0, 1)
print(f"⏳ 请求被限流(429),等待 {wait_time:.2f} 秒后重试..."
f" (第 {attempt + 1}/{max_retries} 次)")
time.sleep(wait_time)
elif response.status_code == 418:
# IP被封禁,不要重试
return {
"success": False,
"error": "IP已被永久封禁(418)",
"retry_after": None
}
elif 500 <= response.status_code < 600:
# 服务器错误,可以重试
wait_time = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"⚠️ 服务器错误({response.status_code}),"
f"{wait_time:.2f}秒后重试...")
time.sleep(wait_time)
else:
# 其他错误,不重试
return {
"success": False,
"error": f"HTTP {response.status_code}",
"retry_after": None
}
except requests.exceptions.Timeout:
wait_time = base_delay * (2 ** attempt)
print(f"⏳ 请求超时,{wait_time:.2f}秒后重试...")
time.sleep(wait_time)
except requests.exceptions.RequestException as e:
return {"success": False, "error": str(e)}
return {
"success": False,
"error": f"已达最大重试次数({max_retries})"
}
实战使用示例
result = exponential_backoff_request(
url="https://api.binance.com/api/v3/order",
headers={"X-MBX-APIKEY": "YOUR_API_KEY"},
params={"symbol": "BTCUSDT", "orderId": 12345}
)
if result["success"]:
print("✅ 请求成功:", result["data"])
else:
print("❌ 请求失败:", result["error"])
实战:构建一个健壮的加密货币行情采集器
现在让我们把以上策略整合起来,构建一个完整的行情采集器。这个采集器会同时从多个交易所获取数据,并优雅地处理各种限流情况。
import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import Dict, List, Optional
from enum import Enum
class Exchange(Enum):
BINANCE = "binance"
OKX = "okx"
BYBIT = "bybit"
@dataclass
class RateLimitConfig:
"""各交易所限流配置"""
requests_per_second: float
burst_size: int
retry_after_seconds: int = 60
class CryptoDataCollector:
"""加密货币行情数据采集器"""
def __init__(self):
self.exchange_configs: Dict[Exchange, RateLimitConfig] = {
Exchange.BINANCE: RateLimitConfig(
requests_per_second=20, # 每秒20个请求
burst_size=100,
retry_after_seconds=60
),
Exchange.OKX: RateLimitConfig(
requests_per_second=10,
burst_size=20,
retry_after_seconds=120
),
Exchange.BYBIT: RateLimitConfig(
requests_per_second=20,
burst_size=100,
retry_after_seconds=60
),
}
# 令牌桶实例
self.buckets: Dict[Exchange, TokenBucket] = {}
for exchange, config in self.exchange_configs.items():
self.buckets[exchange] = TokenBucket(
capacity=config.burst_size,
refill_rate=config.requests_per_second
)
# 统计信息
self.stats = {
"total_requests": 0,
"successful": 0,
"rate_limited": 0,
"errors": 0
}
async def fetch_ticker(self, session: aiohttp.ClientSession,
exchange: Exchange, symbol: str) -> Optional[dict]:
"""获取单个交易对的行情数据"""
bucket = self.buckets[exchange]
# 尝试获取令牌(最多等待5秒)
if not bucket.acquire(tokens=1, timeout=5):
print(f"⚠️ [{exchange.value}] {symbol} 请求超时:限流器已满")
self.stats["rate_limited"] += 1
return None
self.stats["total_requests"] += 1
# 构建请求URL
urls = {
Exchange.BINANCE: f"https://api.binance.com/api/v3/ticker/24hr",
Exchange.OKX: f"https://www.okx.com/api/v5/market/ticker?instId={symbol}",
Exchange.BYBIT: f"https://api.bybit.com/v5/market/tickers?category=spot&symbol={symbol}",
}
try:
async with session.get(urls[exchange],
params={"symbol": symbol} if exchange == Exchange.BINANCE else {},
timeout=aiohttp.ClientTimeout(total=10)) as resp:
if resp.status == 429:
# 被限流,读取Retry-After头
retry_after = int(resp.headers.get('Retry-After', 60))
print(f"🚫 [{exchange.value}] 被限流,{retry_after}秒后恢复")
await asyncio.sleep(retry_after)
self.stats["rate_limited"] += 1
return None
if resp.status == 200:
data = await resp.json()
self.stats["successful"] += 1
return {
"exchange": exchange.value,
"symbol": symbol,
"data": data,
"timestamp": time.time()
}
else:
print(f"❌ [{exchange.value}] HTTP {resp.status}")
self.stats["errors"] += 1
return None
except asyncio.TimeoutError:
print(f"⏱️ [{exchange.value}] {symbol} 请求超时")
self.stats["errors"] += 1
return None
except Exception as e:
print(f"💥 [{exchange.value}] {symbol} 异常: {e}")
self.stats["errors"] += 1
return None
async def collect_all(self, symbols: List[str]) -> List[dict]:
"""从所有交易所采集多个交易对的数据"""
async with aiohttp.ClientSession() as session:
tasks = []
# 为每个交易对在每个交易所创建采集任务
for symbol in symbols:
for exchange in Exchange:
tasks.append(self.fetch_ticker(session, exchange, symbol))
# 并发执行,但受限于各交易所的令牌桶
results = await asyncio.gather(*tasks, return_exceptions=True)
# 过滤掉None和异常
return [r for r in results if r is not None and not isinstance(r, Exception)]
def get_stats(self) -> dict:
"""获取采集统计"""
total = self.stats["total_requests"]
if total > 0:
success_rate = self.stats["successful"] / total * 100
else:
success_rate = 0
return {
**self.stats,
"success_rate": f"{success_rate:.1f}%"
}
使用示例
async def main():
collector = CryptoDataCollector()
# 需要采集的交易对
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
print("📡 开始采集行情数据...")
results = await collector.collect_all(symbols)
print("\n📊 采集统计:")
for key, value in collector.get_stats().items():
print(f" {key}: {value}")
print(f"\n✅ 成功获取 {len(results)} 条数据")
for r in results[:3]: # 显示前3条
print(f" {r['exchange']} {r['symbol']}: {r['data']}")
运行
asyncio.run(main())
常见报错排查
在实战过程中,我整理了开发者最容易遇到的 8 个报错场景及其解决方案。遇到问题不要慌,先看看是不是下面这些情况。
报错 1:HTTP 429 Too Many Requests
错误表现:请求被拒绝,返回 429 状态码,响应体包含 "Too many requests" 或类似信息。
原因分析:你的请求频率超过了交易所的限额。这是最常见的限流错误。
解决方案:
# 检查响应头中的 Retry-After
if response.status_code == 429:
retry_after = int(response.headers.get('Retry-After', 60))
print(f"被限流,需要等待 {retry_after} 秒")
time.sleep(retry_after)
# 如果没有 Retry-After 头,使用指数退避
# 参考上面的指数退避代码
报错 2:IP 被临时封禁(HTTP 418 或持续 429)
错误表现:即使降低请求频率,仍然持续收到 429 错误,或者直接返回 418 状态码。
原因分析:你的 IP 可能被识别为异常流量,进入了临时封禁名单。通常发生在短时间内大量请求,或使用了被标记的 IP 段(如数据中心 IP)。
解决方案:
# 方案一:使用代理 IP 池轮换
proxies = [
"http://proxy1:port",
"http://proxy2:port",
"http://proxy3:port",
]
方案二:使用住宅代理(推荐用于交易)
方案三:联系交易所申请提高限流配额
方案四:检测封禁状态,触发冷却
def check_ip_ban(session):
"""检测IP是否被封禁"""
test_url = "https://api.binance.com/api/v3/ping"
try:
resp = session.get(test_url, timeout=5)
if resp.status_code == 418:
print("🚫 IP已被永久封禁,需要更换IP或联系交易所")
return True
return False
except:
return True # 异常情况下假设被封
报错 3:Request timestamp expired(时间戳过期)
错误表现:签名请求返回 -1021 或 "Timestamp for this request was 1000ms ahead of server's time"。
原因分析:你的服务器时间与交易所服务器时间不同步,差异超过允许范围(Binance 通常是 ±5 秒)。
解决方案:
import time
from datetime import datetime, timezone
def sync_server_time(exchange: str) -> float:
"""同步交易所服务器时间并返回时间偏移量"""
# 获取本地时间戳(毫秒)
local_time_ms = int(time.time() * 1000)
# 发送请求获取服务器时间
if exchange == "binance":
resp = requests.get("https://api.binance.com/api/v3/time")
server_time_ms = resp.json()["serverTime"]
elif exchange == "okx":
resp = requests.get("https://www.okx.com/api/v5/public/time")
server_time_ms = int(float(resp.json()["data"][0]["ts"]))
else:
return 0 # 其他交易所类似
# 计算偏移量
offset_ms = server_time_ms - local_time_ms
print(f"时间偏移: {offset_ms}ms")
return offset_ms
使用:发送签名请求时加上偏移量
TIME_OFFSET = sync_server_time("binance")
def create_signed_params(params: dict) -> dict:
"""创建带时间同步的签名参数"""
params["timestamp"] = int(time.time() * 1000) + TIME_OFFSET
# 然后进行签名...
return params
报错 4:Signature verification failed(签名验证失败)
错误表现:返回 -1022 或 "Signature for this request is not valid"。
原因分析:签名算法不正确,或签名内容与请求参数不匹配。常见原因包括参数排序不一致、编码问题、HMAC 计算错误等。
解决方案:
import hmac
import hashlib
from urllib.parse import urlencode
def create_signature(secret_key: str, params: dict) -> str:
"""生成 Binance 风格的 HMAC-SHA256 签名"""
# 1. 按字母顺序排序参数
sorted_params = sorted(params.items())
# 2. URL编码(保持原始顺序)
query_string = urlencode(sorted_params)
# 3. 使用 HMAC-SHA256 生成签名
signature = hmac.new(
secret_key.encode('utf-8'),
query_string.encode('utf-8'),
hashlib.sha256
).hexdigest()
return signature
使用示例
params = {
"symbol": "BTCUSDT",
"side": "BUY",
"type": "LIMIT",
"quantity": 0.001,
"price": 50000,
"timestamp": int(time.time() * 1000)
}
params["signature"] = create_signature("YOUR_SECRET_KEY", params)
注意:签名参数中不要包含 signature 本身
报错 5:Service unavailable(服务不可用)
错误表现:返回 503 或 "Service unavailable",通常伴随 "ADVISORY" 等字样。
原因分析:交易所服务器过载,正在进行维护,或触发了某种保护机制。
解决方案:
# 检测服务可用性并进行优雅降级
async def resilient_request(session, url, max_retries=3):
"""弹性请求:遇到服务不可用时指数退避重试"""
for attempt in range(max_retries):
try:
async with session.get(url) as resp:
if resp.status == 200:
return await resp.json()
elif resp.status == 503:
wait_time = 2 ** attempt + random.uniform(0, 1)
print(f"服务不可用,等待 {wait_time:.1f}秒后重试...")
await asyncio.sleep(wait_time)
elif resp.status == 429:
# 限流处理
retry_after = resp.headers.get('Retry-After', 60)
await asyncio.sleep(int(retry_after))
else:
return None # 其他错误直接返回
except Exception as e:
await asyncio.sleep(2 ** attempt)
return None # 所有重试都失败
进阶技巧:Rate Limit 的监控与告警
光有处理策略还不够,你需要实时监控自己的 API 使用情况,在即将触发限流前就发出告警。下面是一个简单的监控方案:
import time
from dataclasses import dataclass, field
from typing import Deque
@dataclass
class RateLimitMonitor:
"""Rate Limit 监控器"""
exchange: str
max_requests_per_minute: int
warning_threshold: float = 0.8 # 使用率达到80%时告警
history: Deque = field(default_factory=lambda: Deque(maxlen=100))
def record_request(self):
"""记录一次请求"""
self.history.append(time.time())
def get_current_rate(self) -> float:
"""获取当前使用率(0.0 ~ 1.0)"""
now = time.time()
cutoff = now - 60
# 统计最近60秒的请求数
recent_requests = sum(1 for t in self.history if t >= cutoff)
return recent_requests / self.max_requests_per_minute
def should_warn(self) -> tuple[bool, str]:
"""检查是否应该发出警告"""
rate = self.get_current_rate()
if rate >= 1.0:
return True, f"🚨 [{self.exchange}] 已达限流上限!当前: {rate*100:.0f}%"
elif rate >= self.warning_threshold:
return True, f"⚠️ [{self.exchange}] 使用率过高: {rate*100:.0f}%"
return False, f"✅ [{self.exchange}] 正常: {rate*100:.0f}%"
def get_stats(self) -> dict:
"""获取详细统计"""
return {
"exchange": self.exchange,
"current_rate": f"{self.get_current_rate()*100:.1f}%",
"requests_last_minute": len([t for t in self.history
if t >= time.time() - 60]),
"max_capacity": self.max_requests_per_minute
}
使用示例
monitor = RateLimitMonitor(
exchange="Binance",
max_requests_per_minute=1200,
warning_threshold=0.8
)
在发送请求时记录
def monitored_request():
monitor.record_request()
should_warn, message = monitor.should_warn()
if should_warn:
print(message)
# 可以在这里触发告警:发送邮件/钉钉/飞书通知
结合 AI API 使用:HolySheep 的 Rate Limit 最佳实践
说完加密货币交易所的 rate limit,再聊一个相关场景:当你用 AI API 做量化分析时,同样需要处理 rate limit。
我目前在生产环境使用 HolySheep AI 作为统一的 AI API 中转服务,他们家的 rate limit 设计非常合理:
- 汇率优势:¥1 = $1,无损兑换(官方汇率约 ¥7.3 = $1),节省超过 85% 成本
- 国内直连:延迟 < 50ms,不用再为代理发愁
- 主流模型价格:GPT-4.1 $8/MTok、Claude Sonnet 4.5 $15/MTok、Gemini 2.5 Flash $2.50/MTok、DeepSeek V3.2 $0.42/MTok
- 免费额度:注册即送,可以先测试再决定
他们家对 AI API 的 rate limit 处理比交易所简单很多,而且支持微信/支付宝充值,对国内开发者非常友好。如果你同时在跑加密货币策略和 AI 量化分析,建议统一用 HolySheep 管理所有 API 调用。
# HolySheep AI API 调用示例(支持 OpenAI 兼容格式)
import openai
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY", # 替换为你的 HolySheep API Key
base_url="https://api.holysheep.ai/v1" # HolySheep 官方端点
)
发送请求 - 格式完全兼容 OpenAI SDK
response = client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "system", "content": "你是一个专业的加密货币分析师"},
{"role": "user", "content": "分析一下当前 BTC 的市场情绪和技术指标"}
],
max_tokens=1000,
temperature=0.7
)
print(response.choices[0].message.content)
监控使用量(通过响应头)
X-RateLimit-Remaining: 剩余请求次数
X-RateLimit-Reset: 重置时间戳
适合谁与不适合谁
| 场景 | 推荐程度 | 说明 |
|---|---|---|
| 个人量化交易者 | ⭐⭐⭐⭐⭐ | 完全适合,本文所有策略都适用 |
| 加密货币数据采集项目 | ⭐⭐⭐⭐⭐ | 需要合理使用限流策略,避免被封 |
| 交易所套利机器人 | ⭐⭐⭐⭐ | 适合,但对延迟敏感,需精细调优 |
| 高频交易(HFT) | ⭐⭐ | 公开 API 无法满足需求,需申请专业接口 |
| 薅羊毛/爬虫 | ❌ | 违反服务条款,不推荐 |
价格与回本测算
如果你正在考虑购买交易所的专业 API 套餐或使用第三方数据服务,以下是我的成本分析:
| 方案 | 月费成本 | 优点 | 缺点 |
|---|---|---|---|
| 免费公开 API | $0 | 零成本起步 | 限流严格,不适合高频场景 |
| Binance Pro API | $50/月 | 更高限额,优先支持 | 仅限 Binance 单交易所 |
| 专业数据服务(如 CoinGecko Pro) | $80+/月 | 多交易所聚合 | 延迟较高,无交易功能 |
| 自建代理池 + HolySheep | 代理$20 + AI API按量 | 灵活可控,AI 分析能力强 | 需要一定技术能力 |
以 HolySheep 为例,DeepSeek V3.2 模型只要 $0.42/MTok,做一次完整的加密货币情绪分析大约消耗 50K tokens,成本不到 ¥0.15。如果用 GPT-4.1 同样任务约 $0.4,成本差异明显。合理选型可以大幅节省开支。
为什么选 HolySheep
在众多 AI API 中转服务中,我选择 HolySheep 主要基于三个原因:
第一,汇率优势是实打实的。官方 $1 = ¥7.3,而 HolySheep 做到 ¥1 = $1。假设你一个月用 $100 的 API额度,在官方需要花 ¥730,用 HolySheep 只要 ¥100,差了 7 倍多。
第二,国内访问延迟低。之前用其他中转服务,API 响应要 200-500ms,换成 HolySheep 后稳定在 50ms 以内。对实时性要求高的量化策略来说,这个差距直接影响策略收益。
第三,充值方便。支持微信、支付宝直接充值,不用折腾信用卡或虚拟卡。对国内开发者来说,省去很多麻烦。
他们的 注册链接 送免费额度,建议先体验一下速度,再决定要不要长期使用。
总结
Rate Limit 是每一位加密货币开发者必须跨越的坎。回顾全文,核心要点就三个:
- 理解规则:不同交易所有不同限制,先读文档再开发
- 合理策略:令牌桶、滑动窗口、指数退避,按场景选合适方案
- 监控告警:提前发现问题,而不是等到被封才处理
如果你觉得这篇文章有帮助,欢迎收藏备用。做加密货币开发,坑还有很多,欢迎持续关注我的技术博客。