🛡️ 1688 API限流（429错误）解决方案：多AppKey轮询策略（附Python源码）

一、多AppKey轮询解决什么？

⚠️ 合规注意：多AppKey须是同一企业实名下创建的不同应用，用于分散合法调用量，不是规避计费或绕过封禁。

二、Python：多AppKey轮询 + 令牌桶 + 限流重试

# ali1688_multikey_client.py
import hashlib
import time
import requests
import urllib.parse
import threading
from typing import Dict, List, Optional
from itertools import cycle
from datetime import datetime, timedelta


# ────────────────────────────────────────────────────────
# AppKey 配置（在1688开放平台创建多个「自用型应用」获取）
# ────────────────────────────────────────────────────────
APP_KEYS = [
    {"app_key": "YOUR_APP_KEY_1", "app_secret": "YOUR_APP_SECRET_1"},
    {"app_key": "YOUR_APP_KEY_2", "app_secret": "YOUR_APP_SECRET_2"},
    # {"app_key": "YOUR_APP_KEY_3", ...}  # 按需追加
]

ACCESS_TOKEN = "YOUR_ACCESS_TOKEN"   # 订单类接口需传

# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ────────────────────────────────────────────────────────
# 令牌桶（按单Key上限算，轮询后总QPS≈单Key_QPS×len(APP_KEYS)）
# ────────────────────────────────────────────────────────
class TokenBucket:
    def __init__(self, rate: float = 8.0, capacity: int = None):
        self.rate = rate
        self.cap = capacity or int(rate)
        self.tokens = float(self.cap)
        self.last = time.monotonic()

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


class MultiKey1688Client:
    """
    1688 多AppKey轮询Client
    - 自动按 cycle 分配请求到不同AppKey
    - 每个Key独立令牌桶（默认rate=8 < 免费上限10）
    - 遇429/FLOW_CONTROL自动切换下一Key重试(最多重试 len(keys) 次)
    """

    def __init__(self, app_configs: List[Dict], access_token: str = None,
                 per_key_qps: float = 8.0):
        self.apps = app_configs
        self.token = access_token
        self.buckets = {cfg["app_key"]: TokenBucket(rate=per_key_qps)
                        for cfg in app_configs}
        self.key_cycle = cycle(range(len(app_configs)))
        self.lock = threading.Lock()

    # ─────────── 签名 ───────────
    def _sign(self, params: Dict, app_secret: str) -> str:
        filt = 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 filt)
        return hashlib.md5(
            f"{app_secret}{qs}{app_secret}".encode()
        ).hexdigest().upper()

    def _call(self, url: str, method: str, biz: Dict) -> Dict:
        total_keys = len(self.apps)
        tried = 0

        # 从轮询位置开始，依次试每个Key
        while tried < total_keys:
            idx = next(self.key_cycle)
            cfg = self.apps[idx]
            ak, ask = cfg["app_key"], cfg["app_secret"]
            bucket = self.buckets[ak]

            bucket.wait()   # 限速

            api_p = {
                "method": method,
                "app_key": ak,
                "timestamp": str(int(time.time() * 1000)),
                "format": "json",
                "v": "2.0",
                "sign_method": "md5",
            }
            if self.token:
                api_p["session"] = self.token
            api_p["param2" if "param2" in url or method == "alibaba.offer.search"
                        else "param"] = urllib.parse.quote_plus(
                str(biz).replace("'", '"')
            )
            api_p["sign"] = self._sign(api_p, ask)

            try:
                r = requests.get(url, params=api_p, timeout=15)
                r.raise_for_status()
                d = r.json()

                if "error_response" in d:
                    ec = str(d["error_response"].get("code", ""))
                    msg = d["error_response"].get("msg", "")
                    # 限流 → 换下一个Key重试
                    if "FLOW_CONTROL" in ec or ec == "429" or "ISP_FLOW_CONTROL" in msg:
                        tried += 1
                        if tried < total_keys:
                            print(f"⚠️  Key[{ak[:8]}...] 限流，切换下一Key重试({tried}/{total_keys})")
                            continue
                        raise Exception(f"1688 全部Key限流: {ec} {msg}")
                    # 其他业务错误直接抛
                    raise Exception(f"1688 Err[{ec}]: {msg}")

                result_key = [k for k in d if k != "error_response"][0]
                return d[result_key]

            except requests.exceptions.RequestException:
                # 网络异常也尝试下一Key（可选）
                tried += 1
                if tried >= total_keys:
                    raise
                continue

        raise Exception("MultiKey 调用失败：所有Key均不可用")

    # ─────────── 商品搜索 ───────────
    def search_offers(self, keyword: str, page_no: int = 1,
                       page_size: int = 40, price_min=None, price_max=None) -> Dict:
        biz = {"keywords": keyword, "pageNo": page_no,
               "pageSize": min(page_size, 50), "sortType": "booked"}
        if price_min is not None:
            biz["beginPrice"] = str(int(price_min * 100))
        if price_max is not None:
            biz["endPrice"] = str(int(price_max * 100))
        return self._call(
            "https://gw.open.1688.com/openapi/param2/2/alibaba.offer.search/2.0",
            "alibaba.offer.search", biz
        )

    # ─────────── 商品详情 ───────────
    def get_item(self, offer_id: str, fields: str = None) -> Dict:
        biz = {"item_id": offer_id}
        if fields:
            biz["fields"] = fields
        res = self._call(
            "https://gw.open.1688.com/openapi/http/2/1",
            "alibaba.item.get", biz
        )
        return res.get("alibaba_item_get_response", {}).get("item", {})

    # ─────────── 采购订单列表 ───────────
    def list_orders(self, status="waitsellersend",
                     hours_back=48, page=1, sz=50) -> list:
        biz = {
            "orderStatus": status,
            "gmtCreateStart": (datetime.now() - timedelta(hours=hours_back))
                               .strftime("%Y-%m-%d %H:%M:%S"),
            "gmtCreateEnd": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "pageNo": page, "pageSize": sz
        }
        res = self._call("https://gw.open.1688.com/openapi/http/2/1",
                         "alibaba.trade.buyer.list", biz)
        return res.get("alibaba_trade_buyer_list_response", {}
                      ).get("tradeModelList", []) or []


# =========================================================
# 使用示例
# =========================================================
if __name__ == "__main__":
    client = MultiKey1688Client(
        app_configs=APP_KEYS,
        access_token=ACCESS_TOKEN,
        per_key_qps=8   # 每个Key限速8/s，2个Key≈16/s理论峰值
    )

    try:
        # 搜索
        result = client.search_offers("不锈钢保温杯", price_min=15, price_max=60)
        offers = result.get("offers", [])
        total = result.get("totalResult", 0)
        print(f"✅ 搜到 {total} 条，本页 {len(offers)} 条（多Key轮询）")

        if offers:
            detail = client.get_item(
                str(offers[0]["offerId"]),
                fields="title,price,sku_list,pics"
            )
            print(f"   详情: {detail.get('title')} ¥{detail.get('price')}")

        # 订单（需AccessToken）
        # orders = client.list_orders("waitsellersend")
        # print(f"   待发货采购单: {len(orders)} 笔")

    except Exception as e:
        print(f"❌ {e}")

三、创建多AppKey操作步骤

1. 登录 1688开放平台控制台

2. 应用管理 → 创建应用（选「自用型应用」）

3. 重复创建 N 个（建议2~3个，过多增加维护成本）

4. 每个应用分别申请相同接口权限（商品搜索/订单/物流）

5. 把 AppKey / AppSecret填入上方 APP_KEYS列表

💡 企业通常使用 2个AppKey（搜索用Key-A，订单用Key-B，或同功能做轮询），实测可支撑日调用量 50万+ 不触发限流。

四、生产级注意事项

五、一句话总结（面试版）

1688 API遇429 → 先令牌桶限速（QPS≤免费上限）；高并发场景用多AppKey轮询 + 各Key独立令牌桶 + 限流自动切换下一Key重试，理论可用QPS=单Key上限×Key数。基础商品/订单调用仍免费，多Key仅为分散流量合法使用。