1688开放平台的基础API商品搜索alibaba.offer.search、详情alibaba.item.get、订单查询等确实是免费的但默认受QPS限制通常搜索类≈10/s订单类≈20/s超限会返回ISP_FLOW_CONTROL_LIMIT或 HTTP 429。“破”QPS 10 的本质解法有三层客户端自律令牌桶限速在8~9 QPS留余量防触发风控。架构级扩容申请多个同企业实名下的AppKey做轮询分担理论QPS 单Key上限 × N。官方提额在开放平台购买资源包将单Key提至50/100/200 QPS或升级企业应用。以下是集成了令牌桶限速 多Key轮询 限流退避的GuardedClient源码# guarded_1688_client.py 1688开放平台 GuardedClient - 多AppKey轮询突破单Key QPS限制 - 令牌桶单Key限速 (防触发ISP_FLOW_CONTROL_LIMIT) - 捕获限流异常自动CoolDown 指数退避重试 依赖: pip install requests import hashlib import json import time import requests from typing import Dict, List from threading import Lock # ------------------ 配置区 ------------------ APP_KEYS_SECRETS [ # 同一企业实名下创建的多个自用型应用 {key: YOUR_APP_KEY_1, secret: YOUR_APP_SECRET_1}, {key: YOUR_APP_KEY_2, secret: YOUR_APP_SECRET_2}, # {key: ..., secret: ...}, ] GW https://gw.api.1688.com/openapi MAX_QPS_PER_KEY 8 # 单Key保守限速官方≈10留余量 TOKEN_BUCKET_CAP MAX_QPS_PER_KEY # -------------------------------------------- 封装好API供应商demo urlhttps://console.open.onebound.cn/console/?iLex # 单Key令牌桶 class TokenBucket: def __init__(self, rate: float): self.rate rate self.tokens rate self.ts time.monotonic() self.lock Lock() def wait(self): with self.lock: now time.monotonic() self.token min(self.rate, self.tokens (now - self.ts) * self.rate) self.ts now if self.tokens 1: time.sleep((1 - self.tokens) / self.rate 0.005) self.tokens 0 else: self.tokens - 1 # 多Key轮询Guard class Guarded1688Client: def __init__(self, apps: List[Dict], gwGW): self.apps apps self.gw gw self.idx 0 self.idx_lock Lock() # 为每个Key初始化独立令牌桶 self.buckets [TokenBucket(MAX_QPS_PER_KEY) for _ in apps] self.err_streak [0] * len(apps) # 连续限流计数 def _next_app(self) - (int, Dict, TokenBucket): with self.idx_lock: # 简单轮询可改为按err_streak跳过冷却Key i self.idx self.idx (self.idx 1) % len(self.apps) return i, self.apps[i], self.buckets[i] def _sign_1688(self, params: Dict, secret: str) - str: 1688签名: md5(secret sort(KV) secret) - 大写 注意: 1688部分网关用 HmacMD5这里以通用MD5为例实际请对齐文档 filtered sorted( (k, v) for k, v in params.items() if v is not None and str(v).strip() ! and k ! sign ) qs .join(f{k}{v} for k, v in filtered) raw f{secret}{qs}{secret} return hashlib.md5(raw.encode(utf-8)).hexdigest().upper() def safe_call(self, method: str, biz: Dict, token: str None, max_retry5): method: 如 alibaba.item.get biz: 业务参数 for att in range(max_retry): i, app, bucket self._next_app() bucket.wait() # 单Key QPS守护 # 构造1688 param2风格或router风格参数示意按你网关调整 api_params { method: method, app_key: app[key], timestamp: str(int(time.time() * 1000)), # 1688用毫秒 format: json, v: 2.0, sign_method: md5, **biz } if token: api_params[access_token] token api_params[sign] self._sign_1688(api_params, app[secret]) try: # 1688 param2 路径示例: /openapi/param2/1/{method}/{app_key} url f{self.gw}/param2/1/{method}/{app[key]} resp requests.post(url, dataapi_params, timeout15) resp.raise_for_status() d resp.json() # 1688成功通常在顶层错误在 error_response 或 code!200 if error_response in d: err d[error_response] code str(err.get(code, )) msg err.get(msg, ) # 限流判定 if FLOW_CONTROL in msg or code 15 or resp.status_code 429: self.err_streak[i] 1 back min(2 ** self.err_streak[i], 8) print(f⚠️ Key[{i}] 限流冷却 {back}s 后重试...) time.sleep(back) continue raise Exception(f1688[{code}]: {msg}) # 成功重置错误计数 self.err_streak[i] 0 return d except requests.exceptions.HTTPError as e: if 429 in str(e): self.err_streak[i] 1 time.sleep(2 ** att) continue raise raise RuntimeError(1688 API 多次重试仍限流建议降低QPS或增加AppKey数量) # 使用示例 if __name__ __main__: client Guarded1688Client(APP_KEYS_SECRETS) # 查商品详情免费接口受QPS限制 try: result client.safe_call( alibaba.item.get, {item_id: 612345678901, fields: title,price,sku_list,amount_on_sale}, tokenNone # 商品公开字段可不传token ) print(✅ 1688商品数据:, result.get(alibaba_item_get_response, {}).get(result, {}).get(title)) except Exception as e: print(❌, e)关键设计说明多Key轮询通过Guarded1688Client在多个AppKey间分发请求理论上N个Key可将总QPS提升至N × 10。令牌桶限速每个Key独立令牌桶强制单Key ≤ 8~9 QPS避免触碰官方阈值被拉黑。限流自愈捕获FLOW_CONTROL/ 429 后对该Key做指数退避冷却并自动切换下一个Key重试。合规提醒多Key必须是同一企业主体下合法创建的不同应用用于分散合法业务流量非绕过封禁。如果日调用量极大50万次或对实时性要求高首选买官方资源包提单Key QPS至50/100成本远低于维护多应用复杂度多Key方案更适合初创期零成本扛中等并发的场景。如果你愿意我可以帮你把这套GuardedClient改成Redis中心化令牌桶支持分布式多进程调度或者补一份1688订单增量同步alibaba.trade.buyer.list的完整断点续跑脚本​ 吗