🚦《京东API QPS限流与429规避：从2 QPS到50 QPS的企业升级路径》（附Python源码）

京东JOS免费个人应用默认QPS很低（通常 1~2/s），调订单/商品高频会返回 code=16 / ISP_FLOW_CONTROL_LIMIT或 HTTP 429；

企业实名应用免费上限≈5~10/s，购买资源包可提至 20/50/100 QPS。客户端永远要令牌桶限速 + 限流退避重试，这是规避429的唯一正确做法。

一、京东JOS QPS典型值（2026参考）

• 返回 {"error_response":{"code":16,"zh_desc":"ISP_FLOW_CONTROL_LIMIT..."}}

• 或直返 HTTP 429

二、规避429的核心策略

① 令牌桶限速（QPS ≤ 免费上限×0.8）
   ↓
② 捕获 FLOW_CONTROL / 429
   ↓
③ 指数退避重试（1s → 2s → 4s，最多3次）
   ↓
④ 日额度耗尽（accesscontrol.limited*）→ 暂停翻页记录断点

三、Python：令牌桶 + JOS调用 + 自动退避重试

# jd_rate_limit_client.py
"""
京东JOS API Client with:
 - Token Bucket QPS throttle
 - auto-retry on ISP_FLOW_CONTROL_LIMIT / 429
 - daily-quota exhausted warning
依赖: requests  (pip install requests)
"""
import hashlib
import json
import time
import requests
from typing import Dict, Optional
封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex

# ───────────── 令牌桶 ─────────────
class _TokenBucket:
    def __init__(self, rate: float = 4.0, cap=None):
        self.rate = rate
        self.cap = cap or int(rate)
        self.tok = float(self.cap)
        self.ts = time.monotonic()

    def wait(self):
        now = time.monotonic()
        self.tok = min(self.cap, self.tok + (now - self.ts) * self.rate)
        self.ts = now
        if self.tok >= 1:
            self.tok -= 1
            return
        need = (1 - self.tok) / self.rate
        time.sleep(need + 0.005)
        self.tok = 0

封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ───────────── JOS Client ─────────────
class JdThrottledClient:
    GW = "https://api.jd.com/routerjson"

    def __init__(self, ak, ask, qps=4.0, token=None):
        self.ak, self.ask = ak, ask
        self.bucket = _TokenBucket(rate=qps)
        self._access_token = token

    def _sign(self, p: Dict) -> str:
        filt = sorted((k, v) for k, v in p.items()
                       if v is not None and str(v).strip() != '' and k != 'sign')
        qs = ''.join(f"{k}{v}" for k, v in filt)
        return hashlib.md5(f"{self.ask}{qs}{self.ask}".encode()
                          ).hexdigest().upper()

    def _is_flow_control(self, err_str: str) -> bool:
        return any(k in err_str for k in [
            "ISP_FLOW_CONTROL_LIMIT", "FLOW_CONTROL",
            "code=16", "429"
        ])

    def safe_call(self, method: str, biz: Dict,
                   access_token: str = None,
                   max_retry=3) -> Dict:
        """
        带令牌桶 + 限流自动退避
        """
        at = access_token or self._access_token
        for att in range(max_retry):
            self.bucket.wait()

            api_p = {
                "app_key": self.ak, "method": method,
                "timestamp": str(int(time.time())),   # 秒级!
                "format": "json", "v": "2.0", "sign_method": "md5",
                "360buy_param_json": json.dumps(biz, ensure_ascii=False,
                                                 separators=(',', ':'))
            }
            if at:
                api_p["access_token"] = at
            api_p["sign"] = self._sign(api_p)

            try:
                r = requests.post(self.GW, data=api_p, timeout=15)
                r.raise_for_status()
                d = r.json()

                if "error_response" in d:
                    err = d["error_response"]
                    code = str(err.get("code", ""))
                    sub = err.get("sub_code", "")
                    msg = err.get("zh_desc") or err.get("en_desc", "")

                    # ── 限流 ──
                    if self._is_flow_control(str(err)):
                        if att < max_retry - 1:
                            back = 2 ** att
                            print(f"⚠️  JOS QPS限流，{back}s后第{att+1}次重试...")
                            time.sleep(back)
                            continue
                        raise Exception(f"JOS QPS超限(重试耗尽): {msg}")

                    # ── 日额度 ──
                    if "limited-by-app-access-count" in sub:
                        raise DailyQuotaExhausted(
                            "🚨 京东JOS日免费额度耗尽！超量按基础API计费。"
                            "建议：增量同步 / fields过滤 / 申请提额 / 明早续跑"
                        )

                    # 其它业务错
                    raise Exception(f"JOS[{code}]: {msg} sub={sub}")

                # 取响应
                resp_key = method.replace(".", "_") + "_response"
                if resp_key not in d:
                    for k in d:
                        if k.endswith("_response"):
                            resp_key = k
                            break
                return d.get(resp_key, d)

            except requests.exceptions.RequestException as e:
                if att < max_retry - 1 and "429" in str(e):
                    time.sleep(2 ** att)
                    continue
                raise

        raise RuntimeError("JOS call failed after retries")


# ── 自定义异常 ──
class DailyQuotaExhausted(Exception):
    pass


# =========================================================
# 使用示例 — 商品详情 + 订单列表（自动避让QPS）
# =========================================================
if __name__ == "__main__":
    cli = JdThrottledClient(
        ak="YOUR_JD_APP_KEY",
        ask="YOUR_JD_APP_SECRET",
        qps=4,                      # ← 免费上限5时设4；买包50设40
        token="SELLER_ACCESS_TOKEN" # 订单类需传
    )

    try:
        # ① 商品详情（可不传token）
        ware = cli.safe_call(
            "jingdong.ware.read.get",
            {"wareId": "100012345678",
             "fields": "ware_id,title,jd_price,stock_num"}
        ).get("ware", {})
        print("✅ 商品:", ware.get("title"), "¥", ware.get("jd_price"))

        # ② 订单增量（需token）
        # from datetime import datetime,timedelta
        # now=datetime.now(); start=(now-timedelta(minutes=30)).strftime("%Y-%m-%d %H:%M:%S")
        # end=now.strftime("%Y-%m-%d %H:%M:%S")
        # orders = cli.safe_call("jingdong.pop.order.search", {
        #     "start_modified":start,"end_modified":end,
        #     "order_state":"WAIT_SELLER_STOCK_OUT",
        #     "page":1,"page_size":40
        # }, access_token="SELLER_ACCESS_TOKEN)
        # print("订单数:", orders.get("orderTotal"))

    except DailyQuotaExhausted as e:
        print(e)
        print("→ 记录当前page断点，明早从此继续")
    except Exception as e:
        print("❌", e)

四、企业升级路径（2 QPS → 50 QPS）

五、避坑清单

六、面试/方案一句话

京东JOS限流 = 个人应用默认≈1~2 QPS（易429），企业实名免费≈5~10 QPS，购资源包可提至50/100；客户端用令牌桶限速(QPS≤免费上限×0.8) + 捕获 code=16/429 指数退避重试 + 日额度耗尽断点续跑，不靠裸重试硬冲。