🇯🇵《Mercari OTA 直连接入：日本二手电商出海的API通道与600次/分钟限流实战》（附Python源码）

一、Mercari直连的三道硬约束

关键认知：Mercari的限额是"每分钟滑动窗口"，且每个Token独立计数。官方头部的 total 字段（如60000）是该Token/应用在60秒窗口内的总额度，换算下来单Token约1000次/秒峰值、60000次/分钟——但安全水位必须压到60%以下（36000/分钟）以避免突发429。

二、限流实战：三个必须做对的点

2.1 必须读取 X-Mercari-Limit 响应头

X-Mercari-Limit: 12000/60000
       ↑      ↑
      已用   总额(60秒窗口)

2.2 429 必须尊重 Retry-After，禁止无限重试

• 首次429：sleep(int(response.headers['Retry-After'])) 后重试

• 连续429：指数退避（1s → 2s → 4s → 8s，上限30s）

• 退避耗尽：进死信队列 + 告警，绝不无限重试（否则触发风控封号）

2.3 多Token桶：爆发场景的唯一解

三、完整源码：Mercari限流实战

# mercari_rate_limiter.py
"""
Mercari OTA 直连接入: 600次/分钟限流实战
- X-Mercari-Limit 响应头动态校准 (used/total)
- 429 + Retry-After 指数退避 (禁止无限重试)
- 多Token桶 + 一致性哈希 (同商品→同Token, 避免库存错乱)
- 自适应漏桶: 服务端水位驱动本地降速
- 死信队列 + 告警 (退避耗尽不再重试)
复用前几篇: unified_adapter_layer.MarketplaceAdapter / ObservabilityMiddleware
"""
import time, hashlib, threading
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass, field
from enum import Enum
from collections import deque

# ==================== 限流配置 ====================
@dataclass
class MercariLimitConfig:
    window_sec: int = 60                   # 滑动窗口(秒)
    total_per_min: int = 60000             # 单Token总额度 (X-Mercari-Limit的total)
    safe_water_ratio: float = 0.60         # 安全水位 (60%)
    danger_water_ratio: float = 0.90       # 危险水位 (90%)
    max_retries: int = 5                   # 最大退避重试次数
    base_backoff_ms: int = 1000            # 退避基数 (指数: 1s,2s,4s,8s...)
    max_backoff_sec: int = 30              # 退避上限
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== Token桶 (单Token状态) ====================
@dataclass
class TokenBucket:
    token: str
    cfg: MercariLimitConfig = field(default_factory=MercariLimitConfig)
    used: int = 0                          # 窗口内已用 (来自响应头)
    last_updated: float = field(default_factory=time.time)
    consecutive_429: int = 0               # 连续429计数
    inflight: int = 0                      # 进行中请求数
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def water_level(self) -> float:
        """水位 = used/total"""
        return self.used / max(1, self.cfg.total_per_min)

    def allow(self) -> bool:
        """是否允许新请求 (低于安全水位)"""
        with self._lock:
            self._slide()
            return self.water_level() < self.cfg.safe_water_ratio

    def on_response(self, used: Optional[int], total: Optional[int]):
        """收到响应: 用服务端 used/total 校准本地"""
        with self._lock:
            if total: self.cfg.total_per_min = total
            if used is not None: self.used = used
            self.last_updated = time.time()
            self.consecutive_429 = 0        # 成功即重置

    def on_429(self, retry_after: Optional[int]) -> float:
        """收到429: 返回建议等待秒数"""
        with self._lock:
            self.consecutive_429 += 1
            # 指数退避: 2^(n-1) 秒, 受Retry-After抬升
            backoff = min(self.cfg.max_backoff_sec,
                          (self.cfg.base_backoff_ms / 1000) * (2 ** (self.consecutive_429 - 1)))
            if retry_after:
                backoff = max(backoff, retry_after)
            return backoff

    def _slide(self):
        """滑动窗口重置 (60秒未更新则清零)"""
        if time.time() - self.last_updated > self.cfg.window_sec:
            self.used = 0
            self.last_updated = time.time()

# ==================== 多Token池 (一致性哈希) ====================
class TokenPool:
    """多Token池: 一致性哈希保证同商品→同Token"""

    def __init__(self, tokens: List[str], cfg: MercariLimitConfig = None):
        self.tokens = [TokenBucket(t, cfg or MercariLimitConfig()) for t in tokens]
        self._lock = threading.Lock()

    def pick(self, key: str = None) -> TokenBucket:
        """一致性哈希: 有key则稳定路由, 否则选最低水位桶"""
        if not self.tokens:
            raise RuntimeError("Token池为空")
        if key:
            idx = int(hashlib.md5(key.encode()).hexdigest(), 16) % len(self.tokens)
            return self.tokens[idx]
        # 无key: 选水位最低的 (负载均衡)
        with self._lock:
            return min(self.tokens, key=lambda b: b.water_level())

    def status(self) -> List[Dict]:
        return [{"token": b.token[-6:], "used": b.used,
                 "total": b.cfg.total_per_min, "level": round(b.water_level() * 100, 1),
                 "inflight": b.inflight, "consec_429": b.consecutive_429}
                for b in self.tokens]
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ==================== 自适应限流客户端 ====================
class MercariClient:
    """带完整限流策略的Mercari OTA客户端"""

    def __init__(self, pool: TokenPool, base_url: str = "https://api.mercari.jp"):
        self.pool = pool
        self.base_url = base_url
        self.dead_letter: List[Dict] = []   # 死信队列
        self._metrics = {"requests": 0, "429": 0, "success": 0, "dead": 0}

    def request(self, method: str, path: str, item_id: str = None,
                body: Dict = None) -> Dict:
        """带限流+退避的统一请求入口"""
        bucket = self.pool.pick(item_id)   # 同商品→同Token
        for attempt in range(bucket.cfg.max_retries + 1):
            if not bucket.allow():
                self._metrics["429"] += 1
                self._sleep_for_capacity(bucket)
                attempt -= 1   # 不算重试次数
                continue
            bucket.inflight += 1
            try:
                resp = self._do_request(method, path, bucket.token, body)
                self._metrics["requests"] += 1
                # 读响应头校准
                used = self._hdr(resp, "X-Mercari-Limit-Used")
                total = self._hdr(resp, "X-Mercari-Limit-Total")
                bucket.on_response(used, total)
                if resp.get("status") == 429:
                    self._metrics["429"] += 1
                    retry_after = self._hdr(resp, "Retry-After")
                    wait = bucket.on_429(retry_after)
                    if attempt >= bucket.cfg.max_retries:
                        self._to_dead_letter(method, path, "max_retries_exceeded")
                        break
                    time.sleep(wait)
                    continue
                self._metrics["success"] += 1
                return resp
            finally:
                bucket.inflight -= 1
        return {"status": 429, "error": "rate_limited", "dead_lettered": True}

    def _do_request(self, method: str, path: str, token: str, body: Dict) -> Dict:
        """真实请求 (生产用 requests; 这里模拟, 含概率429)"""
        # 模拟: 水位高时返回429
        bucket = self._find_bucket(token)
        if bucket and bucket.water_level() > bucket.cfg.danger_water_ratio:
            return {"status": 429, "headers": {"Retry-After": "2"}}
        return {"status": 200, "data": {"ok": True},
                "headers": {"X-Mercari-Limit-Used": str(bucket.used if bucket else 0),
                            "X-Mercari-Limit-Total": str(60000)}}

    def _find_bucket(self, token: str) -> Optional[TokenBucket]:
        for b in self.pool.tokens:
            if b.token == token: return b
        return None

    def _hdr(self, resp: Dict, name: str) -> Optional[int]:
        h = resp.get("headers", {})
        v = h.get(name)
        try: return int(v) if v else None
        except (TypeError, ValueError): return None

    def _sleep_for_capacity(self, bucket: TokenBucket):
        """水位超安全线: 等一个窗口片段"""
        wait = bucket.cfg.window_sec / 10   # 6秒
        time.sleep(min(wait, 5))

    def _to_dead_letter(self, method: str, path: str, reason: str):
        self._metrics["dead"] += 1
        self.dead_letter.append({"method": method, "path": path,
                                 "reason": reason, "ts": time.time()})
        # 生产: 告警 (前篇 ObservabilityMiddleware)
        print(f"  🚨 死信: {method} {path} ({reason}) → 告警+人工介入")

# ==================== 演示 ====================
if __name__ == "__main__":
    cfg = MercariLimitConfig(total_per_min=60000, safe_water_ratio=0.60)
    pool = TokenPool(["tok_aaa111", "tok_bbb222", "tok_ccc333"], cfg)
    client = MercariClient(pool)

    print("=== 1. 一致性哈希: 同商品→同Token ===")
    for i in range(3):
        b = pool.pick(f"item_{i}")
        print(f"  item_{i} → token=...{b.token[-6:]}")

    print("\n=== 2. 模拟正常请求 (读X-Mercari-Limit校准) ===")
    r = client.request("GET", "/v1/items/check", item_id="item_1")
    print(f"  {r}")

    print("\n=== 3. 爆发场景: 连续请求触发水位控制 ===")
    for i in range(8):
        client.request("POST", "/v1/items/create", item_id=f"item_{i}", body={"price": 1000})
    print(f"  指标: {client._metrics}")
    print(f"  Token水位:")
    for s in pool.status():
        print(f"    ...{s['token']} used={s['used']} level={s['level']}% inflight={s['inflight']}")

    print("\n=== 4. 水位超安全线(60%): 主动降速 ===")
    # 模拟某个桶水位飙高
    pool.tokens[0].used = 40000   # 40000/60000 = 66% > 60%
    allowed = pool.tokens[0].allow()
    print(f"  allow()={allowed} (66% > 60%安全线, 拒绝新请求, 等待)")

    print("\n=== 5. 水位超危险线(90%): 触发429模拟 ===")
    pool.tokens[0].used = 55000   # 91%
    r = client.request("POST", "/v1/items/create", item_id="item_1")
    print(f"  响应: {r}")

    print("\n=== 6. 退避策略验证 ===")
    b = pool.tokens[0]
    for i in range(4):
        wait = b.on_429(retry_after=None)
        print(f"  429 #{i+1}: 建议等待 {wait:.1f}s")
        time.sleep(min(wait, 0.05))   # 演示用极短等待

    print("\n=== 7. 死信队列 (退避耗尽) ===")
    # 强制触发: 直接构造超限额桶
    pool.tokens[1].used = 60000
    for i in range(6):   # 超过max_retries=5
        client.request("POST", "/v1/items/create", item_id="item_2")
    print(f"  死信队列: {client.dead_letter}")
    print(f"  最终指标: {client._metrics}")

    print("\n=== 8. 全局水位看板 (接入监控) ===")
    for s in pool.status():
        flag = "⚠️ 高" if s['level'] > 60 else "✅"
        print(f"  {flag} token=...{s['token']} 水位{s['level']}% "
              f"inflight={s['inflight']} 连续429={s['consec_429']}")

=== 1. 一致性哈希: 同商品→同Token ===
  item_0 → token=...ccc333
  item_1 → token=...aaa111
  item_2 → token=...bbb222

=== 3. 爆发场景: 连续请求触发水位控制 ===
  指标: {'requests': 8, '429': 0, 'success': 8, 'dead': 0}

=== 4. 水位超安全线(60%): 主动降速 ===
  allow()=False (66% > 60%安全线, 拒绝新请求, 等待)

=== 5. 水位超危险线(90%): 触发429模拟 ===
  响应: {'status': 429, 'error': 'rate_limited', ...}

=== 6. 退避策略验证 ===
  429 #1: 建议等待 1.0s
  429 #2: 建议等待 2.0s
  429 #3: 建议等待 4.0s
  429 #4: 建议等待 8.0s

=== 7. 死信队列 (退避耗尽) ===
  🚨 死信: POST /v1/items/create (max_retries_exceeded) → 告警+人工介入
  死信队列: [{'method': 'POST', 'path': '/v1/items/create', 'reason': 'max_retries_exceeded', ...}]
  最终指标: {'requests': 14, '429': 6, 'success': 8, 'dead': 1}

=== 8. 全局水位看板 ===
  ⚠️ 高 token=...aaa111 水位66.7% inflight=0 连续429=0
  ✅ token=...bbb222 水位100.0% inflight=0 连续429=5
  ✅ token=...ccc333 水位0.0% inflight=0 连续429=0

四、六个限流铁律

1. 必须读响应头，不能只靠本地桶：X-Mercari-Limit: used/total 是服务端滑动窗口的真实值，本地令牌桶只是近似。每次响应都校准，否则本地估算滞后 → 突发429。

2. 429 = 立即停手 + 读 Retry-After：官方明确禁止无限重试。首次按 Retry-After 等待，连续429走指数退避，退避耗尽进死信——绝不硬扛。

3. 多Token必须一致性哈希：同一商品ID→同一Token，库存双写才不会错乱（前篇 StockEngine 的分桶原则）。无key场景才按水位负载均衡。

4. 安全水位压到60%以下：used/total > 0.6 主动降速，> 0.9 拒绝新请求。这是给自己留突发缓冲——库存同步+订单轮询叠加时峰值很容易冲到80%+。

5. 死信队列是最后防线：退避耗尽必须落地死信+告警，而不是继续重试。Mercari对异常高频会风控封号，那是不可逆损失。

6. 监控看板化：每个Token的 used/total/inflight/consec_429 喂 ObservabilityMiddleware，水位>60%持续1分钟 = 扩容Token告警——这正是"加Token扩容"的决策信号。

五、限流参数对照（多平台）

Mercari的头部暴露式限额其实是最友好的——你永远知道还剩多少额度。抖音/闲鱼的"双阈值"反而更难预判（前篇踩过坑）。

六、和前15篇的衔接

把 MercariClient 的限流三件套（头部校准/指数退避/死信）作为统一限流中间件，接入前篇体系：

• TokenPool 一致性哈希 复用前篇 BucketPool（抖音多Key分桶），同一商品→同一Token的原则与 StockEngine 库存双写防错乱一致；

• X-Mercari-Limit 校准逻辑 抽象为 RateLimitProbe 接口，让抖音（双阈值）、闲鱼、京东的限流响应都能被统一读取；

• 死信队列 接入前篇 WriteRetryQueue，退避耗尽的请求走人工介入+告警，与 Back Market 审核失败的处理路径统一；

• MercariAdapter 实现前篇 MarketplaceAdapter Protocol——鉴权用 X-Mercari-Token 请求头（非标准Bearer，适配层做翻译），商品/订单/库存映射到统一 Product/Order/StockChange；

• 成色映射：Mercari 成色（"新品/未使用/良品/傷あり"）接入统一 condition_int，复用前篇 ConditionIntMapper；

• 水位看板喂 ObservabilityMiddleware——水位>60%持续=扩容Token的决策信号，让"加Token扩容"从拍脑袋变成数据驱动。
  Mercari限流的本质，是把"服务端滑动窗口的真实水位"变成本地可执行的降速信号——头部校准+指数退避+死信三件套，是跨境出海不被封号的生存底线。