当前位置：首页 > news >正文

《1688开放平台基础API真免费？QPS 10的限制怎么破（附GuardedClient源码）》

news 2026/7/18 9:44:14

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 url=https://console.open.onebound.cn/console/?i=Lex # ===== 单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], gw=GW): 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_retry=5): """ 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, data=api_params, timeout=15) 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(f"1688[{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"}, token=None # 商品公开字段可不传token ) print("✅ 1688商品数据:", result.get("alibaba_item_get_response", {}).get("result", {}).get("title")) except Exception as e: print("❌", e)