共计 2067 个字符，预计需要花费 6 分钟才能阅读完成。
背景痛点
直接调用 OpenAI 官方 API 时，开发者常遇到三个典型问题：
- 地域限制 ：部分国家 / 地区的 IP 可能被屏蔽，导致服务不可用
- 计费模型 ：按 token 计费方式在持续交互场景下成本难以控制
- QPS 瓶颈 ：免费账户每分钟仅有 3 次请求配额，商业版也有并发限制
技术选型
| 方案 | 平均延迟 | 成本 | 维护性 | 适用场景 |
|---|---|---|---|---|
| Nginx 反向代理 | 80-120ms | 低 | 中等 | 自有服务器 + 长期使用 |
| Cloudflare Workers | 200-300ms | 免费额度 | 简单 | 快速验证 + 无服务器架构 |
核心实现
Docker 部署方案
# docker-compose.yml
services:
nginx:
image: nginx:1.25
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
- ./cache:/var/cache/nginx
ports:
- "8000:8000"
environment:
- OPENAI_ENDPOINT=${OPENAI_ENDPOINT}关键 Nginx 配置
location /v1/chat/completions {proxy_pass ${OPENAI_ENDPOINT};
proxy_set_header Authorization "Bearer ${API_KEY}";
proxy_set_header Content-Type "application/json";
# 负载均衡
proxy_next_upstream error timeout invalid_header http_500;
# 请求伪装
proxy_set_header User-Agent "${RANDOM_UA}";
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}令牌桶限流算法
class TokenBucket:
def __init__(self, capacity: int, fill_rate: float):
self.capacity = capacity
self.tokens = capacity
self.last_fill = time.time()
self.fill_rate = fill_rate
def consume(self, tokens=1) -> bool:
now = time.time()
elapsed = now - self.last_fill
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.fill_rate
)
self.last_fill = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False避坑指南
- UA 轮换策略 ：维护常见浏览器 UA 列表，每次请求随机选择
- 429 降级方案 ：触发限流时返回缓存的最近成功响应
- 敏感信息过滤 ：
/("api_key":\s*")([^"]+)/gi # 替换为 $1[REDACTED]
性能测试
使用 Locust 进行压测（100 并发用户）：
| 指标 | 直连 API | 镜像服务 |
|---|---|---|
| 平均 TPS | 42 | 118 |
| 95% 延迟 (ms) | 2100 | 650 |
Prometheus 监控配置片段：
- job_name: 'nginx'
static_configs:
- targets: ['nginx:9113']
metrics_path: /stub_status安全加固
JWT 鉴权中间件
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
token = request.headers.get("Authorization")
try:
payload = jwt.decode(token.split(" ")[1],
key=os.getenv("JWT_SECRET"),
algorithms=["HS256"]
)
request.state.user = payload["sub"]
except Exception as e:
return JSONResponse(
status_code=401,
content={"detail": "Invalid token"}
)
return await call_next(request)日志脱敏规则
import re
def sanitize_log(text: str) -> str:
patterns = [(r"(sk-)[a-zA-Z0-9]{24}", "[API_KEY]"),
(r"\b[\w.]+@[\w.]+\b", "[EMAIL]")
]
for pattern, replacement in patterns:
text = re.sub(pattern, replacement, text)
return text开放问题
当用户量增长到需要跨多台服务器部署时，传统的单机限流方案会失效。你会如何设计分布式限流系统来保证全局的速率控制？考虑使用 Redis+Lua 脚本实现跨节点计数，或者采用分层限流架构？欢迎在评论区分享你的设计方案。
正文完
