速率限制

Quick Reference

WhatA55 API 速率限制策略和重试策略

Why避免在高流量期间出现服务中断

Reading Time7 分钟

Difficulty中级

Prerequisites可运行的 API 集成

策略

参数	值
速率限制	每个 API（应用程序编程接口）密钥每分钟 100 个请求
窗口	滚动 60 秒窗口
范围	所有端点按 client_id 计算
达到限制时的响应	HTTP 429 Too Many Requests

速率限制头部

每个 API 响应都包含速率限制头部：

HTTP/1.1 200 OK
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 73
X-RateLimit-Reset: 1711036800
Content-Type: application/json

头部	描述
X-RateLimit-Limit	每个窗口允许的最大请求数
X-RateLimit-Remaining	当前窗口剩余请求数
X-RateLimit-Reset	窗口重置的 Unix 时间戳

HTTP 429 响应

当您超过限额时：

{
  "error": {
    "code": "RATE_LIMIT_EXCEEDED",
    "message": "Too many requests. Retry after 23 seconds.",
    "details": [
      {
        "field": "rate_limit",
        "reason": "100 requests per minute exceeded"
      }
    ]
  },
  "request_id": "req_abc123..."
}

Retry-After 头部指示需要等待的秒数：

HTTP/1.1 429 Too Many Requests
Retry-After: 23
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1711036823

指数退避重试策略

cURL

#!/bin/bash
MAX_RETRIES=3
RETRY_DELAY=1

for i in $(seq 1 $MAX_RETRIES); do
RESPONSE=$(curl -s -w "\n%{http_code}" \
  -X POST "https://sandbox.api.a55.tech/api/v1/bank/wallet/charge/" \
  -H "Authorization: Bearer ${A55_ACCESS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"amount": "100.00", "currency": "BRL"}')

HTTP_CODE=$(echo "$RESPONSE" | tail -1)
BODY=$(echo "$RESPONSE" | head -n -1)

if [ "$HTTP_CODE" -ne 429 ] && [ "$HTTP_CODE" -lt 500 ]; then
  echo "$BODY"
  exit 0
fi

echo "第 $i 次尝试失败（HTTP $HTTP_CODE）。将在 ${RETRY_DELAY} 秒后重试..."
sleep $RETRY_DELAY
RETRY_DELAY=$((RETRY_DELAY * 2))
done

echo "所有重试次数已用尽"
exit 1

Python

import os
import time
import random
import requests

def request_with_backoff(method, url, max_retries=3, **kwargs):
  headers = {
      "Authorization": f"Bearer {os.environ['A55_ACCESS_TOKEN']}",
      **kwargs.pop("headers", {}),
  }

  for attempt in range(max_retries + 1):
      try:
          response = requests.request(method, url, headers=headers, **kwargs)

          remaining = response.headers.get("X-RateLimit-Remaining")
          if remaining and int(remaining) < 10:
              print(f"警告：窗口内剩余 {remaining} 个请求")

          if response.status_code == 429:
              retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
              jitter = random.uniform(0, 1)
              time.sleep(retry_after + jitter)
              continue

          if response.status_code >= 500:
              delay = (2 ** attempt) + random.uniform(0, 1)
              time.sleep(delay)
              continue

          return response
      except requests.ConnectionError:
          if attempt == max_retries:
              raise
          time.sleep(2 ** attempt)

  raise Exception(f"请求在 {max_retries} 次重试后失败")

Node.js

async function requestWithBackoff(method, url, options = {}, maxRetries = 3) {
const headers = {
  Authorization: `Bearer ${process.env.A55_ACCESS_TOKEN}`,
  "Content-Type": "application/json",
  ...options.headers,
};

for (let attempt = 0; attempt <= maxRetries; attempt++) {
  try {
    const res = await fetch(url, { method, headers, ...options });

    const remaining = res.headers.get("X-RateLimit-Remaining");
    if (remaining && parseInt(remaining) < 10) {
      console.warn(`警告：窗口内剩余 ${remaining} 个请求`);
    }

    if (res.status === 429) {
      const retryAfter = parseInt(res.headers.get("Retry-After") || 2 ** attempt);
      const jitter = Math.random();
      await new Promise(r => setTimeout(r, (retryAfter + jitter) * 1000));
      continue;
    }

    if (res.status >= 500) {
      const delay = (2 ** attempt + Math.random()) * 1000;
      await new Promise(r => setTimeout(r, delay));
      continue;
    }

    return res;
  } catch (err) {
    if (attempt === maxRetries) throw err;
    await new Promise(r => setTimeout(r, 2 ** attempt * 1000));
  }
}

throw new Error(`请求在 ${maxRetries} 次重试后失败`);
}

最佳实践

1. 读取速率限制头部——监控 X-RateLimit-Remaining，在归零之前主动降速
2. 不要重试 4xx 错误（429 除外）——客户端错误表明请求中有 bug，而非暂时性故障
3. 为退避添加抖动——防止多个客户端同时重试时产生惊群效应（thundering herd）
4. 使用请求队列——通过排队请求并以受控速率消费来平滑流量峰值
5. 尽可能使用缓存——缓存 GET 响应（钱包详情、收款状态）以减少请求量
6. 批量操作——如果需要创建多笔收款，将其分散在窗口期内，而非一次性全部发送

幂等性 →在重试时防止重复收款调试技巧 →排查 API 问题常见错误 →避免集成陷阱