《京东API QPS限流与429规避从2 QPS到50 QPS的企业升级路径》附Python源码直接说结论京东JOS免费个人应用默认QPS很低通常1~2/s调订单/商品高频会返回code16 / ISP_FLOW_CONTROL_LIMIT或 HTTP 429企业实名应用免费上限≈5~10/s购买资源包可提至 20/50/100 QPS。客户端永远要令牌桶限速 限流退避重试这是规避429的唯一正确做法。一、京东JOS QPS典型值2026参考应用类型免费QPS(估)买包可提至说明个人开发者1~2/s❌ 不支持买包仅学习/比价企业自用(未买包)5/s(商品) 10/s(订单部分)—需企业支付宝认证企业资源包—20 / 50 / 100 / 200 s按接口族购买京东联盟API5~10/s可申请提备案后超限表现返回{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 urlhttps://console.open.onebound.cn/console/?iLex # ───────────── 令牌桶 ───────────── class _TokenBucket: def __init__(self, rate: float 4.0, capNone): 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 # ───────────── JOS Client ───────────── class JdThrottledClient: GW https://api.jd.com/routerjson def __init__(self, ak, ask, qps4.0, tokenNone): self.ak, self.ask ak, ask self.bucket _TokenBucket(rateqps) 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, code16, 429 ]) def safe_call(self, method: str, biz: Dict, access_token: str None, max_retry3) - 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_asciiFalse, separators(,, :)) } if at: api_p[access_token] at api_p[sign] self._sign(api_p) try: r requests.post(self.GW, dataapi_p, timeout15) 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后第{att1}次重试...) time.sleep(back) continue raise Exception(fJOS QPS超限(重试耗尽): {msg}) # ── 日额度 ── if limited-by-app-access-count in sub: raise DailyQuotaExhausted( 京东JOS日免费额度耗尽超量按基础API计费。 建议增量同步 / fields过滤 / 申请提额 / 明早续跑 ) # 其它业务错 raise Exception(fJOS[{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( akYOUR_JD_APP_KEY, askYOUR_JD_APP_SECRET, qps4, # ← 免费上限5时设4买包50设40 tokenSELLER_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 # nowdatetime.now(); start(now-timedelta(minutes30)).strftime(%Y-%m-%d %H:%M:%S) # endnow.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_tokenSELLER_ACCESS_TOKEN) # print(订单数:, orders.get(orderTotal)) except DailyQuotaExhausted as e: print(e) print(→ 记录当前page断点明早从此继续) except Exception as e: print(❌, e)四、企业升级路径2 QPS → 50 QPS阶段做法效果① 个人→企业​开放平台→企业实名(营业执照)→创建自用型应用免费QPS ↑ 至 5~10/s② 令牌桶保守限速​代码qps4免费5时或qps8免费10时基本不断流③ 购买资源包​应用→流量配额→购 JOS QPS 提升包(20/50/100档)单Key QPS ↑④ 多AppKey轮询(超大并发)​同1688多Key策略N个Key各带令牌桶理论 QPSN×单Key上限⑤ 聚石塔部署​ECS放聚石塔内基础API单价更低稳定性更好五、避坑清单坑现象解决个人应用频繁429默认1~2/s切企业应用 或 极度降速(qps1)timestamp毫秒Invalid TimestampJOS用秒级​int(time.time())订单接口403但商品OK个人号无订单权限企业应用卖家OAuth token翻页超日额度limited-by-app-access-count增量按 modified 断点续跑买包仍429新包未生效 / 调错接口族控制台确认生效接口含当前method六、面试/方案一句话京东JOS限流 个人应用默认≈1~2 QPS易429企业实名免费≈5~10 QPS购资源包可提至50/100客户端用令牌桶限速(QPS≤免费上限×0.8) 捕获 code16/429 指数退避重试 日额度耗尽断点续跑不靠裸重试硬冲。需要我补多AppKey轮询提QPS的完整示例​ 或APScheduler 京东订单增量同步断点续跑Token刷新​ 吗