В Люберцах мы убираем неопределённость у...
注册账号免费测试1688API接口1688 的 item_review 接口（官方标准方法名：alibaba.item.review.get）是阿里巴巴开放平台提供的核心接口之一，用于获取指定商品的买家评论数据，包括文本评价、评分、晒图、采购属性（如采购数量、...
还有 80 人发表了评论 加入10618人围观未分类
一套接口打三平台，省下 80% 的对接工时（附 Python 源码）
万邦科技Lex 发表于2026-10-23 浏览8 评论0
# 一套接口打三平台，省下 80% 的对接工时（附 Python 源码）
> 80% 不是省在第一次对接，是省在之后每一次改动。这笔账要算清楚，否则你会以为统一接口层是过度设计。
## 一、三套代码的日常
先看一个很常见的项目结构：
```
project/
├── taobao/
│ ├── client.py # 签名 + 请求
│ ├── product.py # 商品：字段映射、SKU 处理
│ ├── order.py # 订单：分页、状态转换
│ └── logistics.py
├── alibaba/
│ ├── client.py # 又是一份签名
│ ├── product.py # 又是一份字段映射
│ ├── order.py
│ └── logistics.py
├── jd/
│ ├── client.py
│ ├── product.py
│ ├── order.py
│ └── logistics.py
└── service/
├── price_sync.py # ← 业务逻辑，里面全是 if platform == "taobao"
├── stock_sync.py # ← 又一次 if platform == ...
└── order_sync.py # ← 第三次
```
问题不在文件多，在于**每个业务需求都要改三处**。
产品说了句「库存要精确到 SKU」，你得改三个 `stock` 模块。发现淘宝有个坑要修，另外两个平台得确认一遍是不是也有同样的问题。加个新字段，三个平台各写一遍映射。
代码量翻三倍，bug 面积也翻三倍——因为三份实现里总有一份写得跟另外两份不一样。
## 二、先算一笔诚实的账
「省 80% 工时」这种话，如果只算首次对接，是站不住的。算细一点：
### 首次对接
| 项目 | 三套独立实现 | 统一接口层 |
| --------- | ------------- | ----------------- |
| 签名/请求基础设施 | 3 × 1 天 = 3 天 | 1 天（抽象基类） |
| 商品资源适配 | 3 × 2 天 = 6 天 | 3 × 1.5 天 = 4.5 天 |
| 订单资源适配 | 3 × 2 天 = 6 天 | 3 × 1.5 天 = 4.5 天 |
| 物流资源适配 | 3 × 1 天 = 3 天 | 3 × 0.5 天 = 1.5 天 |
| 统一层设计 | 0 | 1.5 天 |
| **合计** | **18 天** | **13 天** |
首次对接只省 **28%**。因为适配器还是得一个一个写，平台差异不会因为抽象就消失。
### 后续每一次改动
真正拉开差距的是这里。假设一个持续一年的项目：
| 变更类型 | 频率 | 三套实现 | 统一层 |
| --------- | ------- | ------------------------ | ----------- |
| 加一个业务字段 | 每月 1 次 | 3 处 × 0.5 天 = 1.5 天 | 1 处 × 0.5 天 |
| 修一个平台 bug | 每月 2 次 | 2 × 0.5 天（有时要排查是否影响其他两份） | 0.5 天 |
| 调整业务规则 | 每季度 1 次 | 3 处 × 1 天 | 1 处 × 1 天 |
| 接第四个平台 | 每年 1 次 | 5 天 | 1.5 天 |
| 新人接手理解成本 | 每年 1 次 | 3 天 | 1 天 |
一年下来：三套实现约 **43 天**，统一层约 **16 天**。
**维护期省 63%。** 加上首次的 28%，以及「bug 只修一处就不会漏」带来的隐性收益，综合到 80% 是合理的。
关键结论：**统一接口层的价值随时间增长，项目越久省得越多。** 如果只做一个短期的数据导出脚本，确实没必要。但如果是长期跑的业务系统，这个投入一年内就回本。
## 三、架构：门面 + 适配器 + 能力声明
```
业务代码（只写一次）
↓
CommerceAPI（门面）
├─ 统一重试 / 缓存 / 限流
├─ 能力检查（不支持就明确报错）
└─ 结果归一化
↓
┌─────────┼─────────┐
TaobaoAdapter AlibabaAdapter JdAdapter
↓ ↓ ↓
各自签名、字段、分页、错误码
```
三个角色的职责必须分清：
- **门面**：业务逻辑唯一入口，负责跨平台共性（重试、缓存、能力检查）
- **适配器**：只做「平台方言 → 统一语言」的翻译，不含业务判断
- **能力声明**：显式声明每个平台支持什么，不支持的操作直接抛错
第三条是最容易被忽略、也最容易埋雷的。下面单独讲。
## 四、代码
### 4.1 统一模型和能力声明
```python
# core/models.py
from dataclasses import dataclass, field
from decimal import Decimal
@dataclass
class Product:
platform: str
item_id: str
title: str
price: Decimal
stock: int | None = None # None 表示该平台不提供
images: list[str] = field(default_factory=list)
skus: list["Sku"] = field(default_factory=list)
@dataclass
class Sku:
sku_id: str
props: dict[str, str]
price: Decimal
stock: int | None = None
@dataclass
class Order:
platform: str
order_id: str
status: str # 统一状态，不是平台原始值
pay_amount: Decimal
created_at: str
modified_at: str
buyer: str = ""
@dataclass
class Capability:
"""
每个平台支持什么，显式声明。
这是统一接口里最关键的一层——三平台能力根本不对等，
不声明清楚，业务代码就会在某个平台上拿到空数据还以为正常。
"""
sku_level_stock: bool = False # 库存是否精确到 SKU
tiered_price: bool = False # 是否有阶梯价（1688 特有）
order_increment: bool = False # 订单是否支持按修改时间增量拉
logistics_subscribe: bool = False # 物流是否支持订阅推送
batch_product_query: bool = False # 商品是否支持批量查询
```
为什么不用 `hasattr` 或者 try/except 判断？因为**能力应该是显式契约，不是运行时探针**。业务代码在启动时就能知道「京东这个接口不支持 SKU 级库存」，从而决定降级策略，而不是等跑到一半发现返回全是 `None`。
### 4.2 适配器基类
```python
# core/adapter.py
from abc import ABC, abstractmethod
from decimal import Decimal
from .models import Product, Order, Capability
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class NotSupported(Exception):
"""平台明确不支持该操作。不要用假数据糊弄调用方。"""
def __init__(self, platform: str, operation: str):
super().__init__(f"[{platform}] 不支持操作：{operation}")
self.platform = platform
self.operation = operation
class BaseAdapter(ABC):
platform: str = ""
capability: Capability = Capability()
# --- 子类必须实现 ---
@abstractmethod
async def get_product(self, item_id: str) -> Product:
...
@abstractmethod
async def list_orders(self, start: str, end: str, page: int = 1) -> list[Order]:
...
# --- 子类按能力选择性实现 ---
async def get_stock(self, item_id: str, sku_id: str | None = None) -> int:
if not self.capability.sku_level_stock and sku_id:
raise NotSupported(self.platform, "按 SKU 查询库存")
return await self._get_stock_impl(item_id, sku_id)
async def _get_stock_impl(self, item_id: str, sku_id: str | None) -> int:
raise NotSupported(self.platform, "库存查询")
# --- 状态归一：每个平台状态码不同，统一成五个 ---
STATUS_MAP: dict[str, str] = {}
def normalize_status(self, raw: str) -> str:
return self.STATUS_MAP.get(str(raw), "UNKNOWN")
```
统一状态只定五个：`PENDING`（待付款）、`PAID`（待发货）、`SHIPPED`（已发货）、`DONE`（已完成）、`CLOSED`（已关闭）。多了没用，少了不够用。
### 4.3 三个适配器（只展示关键差异）
```python
# adapters/taobao.py
from decimal import Decimal
from core.adapter import BaseAdapter
from core.models import Product, Order, Capability, Sku
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class TaobaoAdapter(BaseAdapter):
platform = "taobao"
capability = Capability(
sku_level_stock=True,
order_increment=True,
logistics_subscribe=True,
batch_product_query=False,
)
STATUS_MAP = {
"WAIT_BUYER_PAY": "PENDING",
"WAIT_SELLER_SEND_GOODS": "PAID",
"WAIT_BUYER_CONFIRM_GOODS": "SHIPPED",
"TRADE_FINISHED": "DONE",
"TRADE_CLOSED": "CLOSED",
}
def __init__(self, client):
self.c = client
async def get_product(self, item_id: str) -> Product:
d = await self.c.call("taobao.item.seller.get", {
"fields": "num_iid,title,price,num,quantity,pic_url,sku",
"num_iid": item_id,
})
it = d["item_get_response"]["item"]
skus = []
for s in (it.get("skus") or {}).get("sku") or []:
skus.append(Sku(
sku_id=str(s["sku_id"]),
props=_parse_props(s.get("properties_name", "")),
price=Decimal(str(s.get("price", "0"))),
stock=int(s.get("quantity", 0)),
))
return Product(
platform=self.platform,
item_id=str(it["num_iid"]),
title=it.get("title", ""),
price=Decimal(str(it.get("price", "0"))),
stock=int(it.get("quantity", 0)),
images=[it["pic_url"]] if it.get("pic_url") else [],
skus=skus,
)
async def list_orders(self, start, end, page=1) -> list[Order]:
d = await self.c.call("taobao.trades.sold.get", {
"fields": "tid,status,payment,created,modified,buyer_nick",
"start_modified": start, "end_modified": end,
"page_no": page, "page_size": 100,
})
rows = ((d.get("trades_sold_get_response") or {})
.get("trades") or {}).get("trade") or []
return [Order(
platform=self.platform,
order_id=str(t["tid"]),
status=self.normalize_status(t.get("status", "")),
pay_amount=Decimal(str(t.get("payment", "0"))),
created_at=t.get("created", ""),
modified_at=t.get("modified", ""),
buyer=t.get("buyer_nick", ""),
) for t in rows]
def _parse_props(s: str) -> dict[str, str]:
out = {}
for seg in filter(None, s.split(";")):
p = seg.split(":")
if len(p) >= 4:
out[p[2]] = p[3]
return out
```
```python
# adapters/alibaba.py
from decimal import Decimal
from core.adapter import BaseAdapter
from core.models import Product, Order, Capability
from core.adapter import NotSupported
class AlibabaAdapter(BaseAdapter):
platform = "alibaba"
capability = Capability(
sku_level_stock=False, # 1688 按起订量阶梯，没有 SKU 概念
tiered_price=True,
order_increment=True,
logistics_subscribe=False, # 需要单独申请
)
STATUS_MAP = {
"waitbuyerpay": "PENDING",
"waitsellersend": "PAID",
"waitbuyerreceive": "SHIPPED",
"success": "DONE",
"cancel": "CLOSED",
}
def __init__(self, client):
self.c = client
async def get_product(self, item_id: str) -> Product:
d = await self.c.call("com.alibaba.product.get", {
"productID": item_id, "webSite": "1688",
})
r = d.get("result") or {}
ranges = (r.get("saleInfo") or {}).get("priceRanges") or []
price = Decimal(str(ranges[0]["price"])) if ranges else Decimal("0")
return Product(
platform=self.platform,
item_id=str(r.get("productID", item_id)),
title=r.get("subject", ""),
price=price,
stock=int((r.get("saleInfo") or {}).get("amountOnSale", 0) or 0),
images=[i["fullPathImageURI"]
for i in (r.get("image") or {}).get("images", [])],
)
async def get_stock(self, item_id: str, sku_id: str | None = None) -> int:
# 显式拒绝，而不是忽略 sku_id 返回总量
if sku_id:
raise NotSupported(self.platform, "按 SKU 查询库存（1688 无此概念）")
return await self._get_stock_impl(item_id, None)
async def list_orders(self, start, end, page=1) -> list[Order]:
d = await self.c.call("com.alibaba.trade/get.soldList", {
"modifyStartTime": start, "modifyEndTime": end,
"pageNo": page, "pageSize": 50,
})
rows = (d.get("result") or {}).get("orderList") or []
return [Order(
platform=self.platform,
order_id=str(o.get("orderId")),
status=self.normalize_status(o.get("orderStatus", "")),
pay_amount=Decimal(str(o.get("sumPayment") or "0")),
created_at=o.get("createTime", ""),
modified_at=o.get("modifyTime", ""),
buyer=o.get("toLoginId", ""),
) for o in rows]
```
```python
# adapters/jd.py
from decimal import Decimal
from core.adapter import BaseAdapter
from core.models import Product, Order, Capability
class JdAdapter(BaseAdapter):
platform = "jd"
capability = Capability(
sku_level_stock=True,
order_increment=True,
logistics_subscribe=True,
)
STATUS_MAP = {
"1": "PENDING", # 等待付款
"2": "PAID", # 等待发货
"3": "SHIPPED", # 已发货
"4": "DONE", # 已完成
"6": "CLOSED", # 已取消
}
def __init__(self, client):
self.c = client
async def get_product(self, item_id: str) -> Product:
d = await self.c.call("jd.union.open.goods.query", {
"360buy_param_json": '{"skuIds":[%s]}' % item_id,
})
rows = ((d.get("jd_union_open_goods_query_responce") or {})
.get("result") or {}).get("data") or []
if not rows:
return Product(platform=self.platform, item_id=item_id,
title="", price=Decimal("0"))
g = rows[0]
return Product(
platform=self.platform,
item_id=str(g.get("skuId", item_id)),
title=g.get("skuName", ""),
price=Decimal(str(g.get("price", "0"))),
stock=None, # 该接口不返回库存
images=[g["imageUrl"]] if g.get("imageUrl") else [],
)
async def list_orders(self, start, end, page=1) -> list[Order]:
d = await self.c.call("jingdong.pop.order.search", {
"360buy_param_json": (
'{"page":"%d","pageSize":"100",'
'"startDate":"%s","endDate":"%s"}' % (page, start, end)),
})
rows = ((d.get("jingdong_pop_order_search_responce") or {})
.get("searchorderinfo_result") or {}).get("orderInfoList") or []
return [Order(
platform=self.platform,
order_id=str(o.get("orderId")),
status=self.normalize_status(o.get("orderState", "")),
pay_amount=Decimal(str(o.get("orderPayment") or "0")),
created_at=o.get("orderStartTime", ""),
modified_at=o.get("orderModified", ""),
buyer=o.get("pin", ""),
) for o in rows]
```
三个适配器放在一起看，共性一目了然：**都要签名、都要分页、都要状态映射、都要字段归一**。这些「一样的东西」全部沉到基类，适配器只保留「不一样的东西」。
这就是统一接口层真正省事的地方——不是省打字，是**让「不一样」变得可枚举**。
### 4.4 门面
```python
# core/api.py
import asyncio
from .adapter import BaseAdapter, NotSupported
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
class CommerceAPI:
"""
业务代码唯一入口。三平台一个门面，业务层不再出现 platform 判断。
"""
def __init__(self, adapters: dict[str, BaseAdapter]):
self.adapters = adapters
def _get(self, platform: str) -> BaseAdapter:
if platform not in self.adapters:
raise KeyError(f"未注册平台：{platform}")
return self.adapters[platform]
# --- 单平台操作 ---
async def product(self, platform: str, item_id: str):
return await self._get(platform).get_product(item_id)
async def orders(self, platform: str, start: str, end: str,
max_pages: int = 50) -> list:
ad = self._get(platform)
if not ad.capability.order_increment:
raise NotSupported(platform, "订单增量拉取")
out, page = [], 1
while page <= max_pages:
batch = await ad.list_orders(start, end, page)
out.extend(batch)
if len(batch) < 100:
break
page += 1
return out
# --- 跨平台操作：这才是统一接口真正的价值 ---
async def products_all(self, item_ids_map: dict[str, list[str]],
concurrency: int = 10) -> dict[str, list]:
"""
一次拿三个平台的商品。业务层不用写循环、不用管并发。
item_ids_map = {"taobao": [...], "alibaba": [...], "jd": [...]}
"""
sem = asyncio.Semaphore(concurrency)
result: dict[str, list] = {}
async def one(platform: str, iid: str):
async with sem:
try:
return platform, await self.product(platform, iid)
except Exception as e:
print(f"[{platform}] {iid} 失败: {e}")
return platform, None
tasks = [one(p, i) for p, ids in item_ids_map.items() for i in ids]
for platform, prod in await asyncio.gather(*tasks):
if prod is not None:
result.setdefault(platform, []).append(prod)
return result
async def order_summary(self, start: str, end: str,
platforms: list[str] | None = None) -> dict:
"""
跨平台订单汇总。业务层写一行，三平台自动并行。
"""
targets = platforms or list(self.adapters.keys())
async def one(p):
try:
orders = await self.orders(p, start, end)
return p, {
"count": len(orders),
"amount": sum(o.pay_amount for o in orders),
"by_status": _group_status(orders),
}
except Exception as e:
return p, {"error": str(e)}
pairs = await asyncio.gather(*[one(p) for p in targets])
return dict(pairs)
def _group_status(orders) -> dict[str, int]:
out: dict[str, int] = {}
for o in orders:
out[o.status] = out.get(o.status, 0) + 1
return out
```
`order_summary` 这个方法值得多看一眼。在没统一接口的项目里，这段逻辑要写三遍，每遍处理不同的分页、不同的状态码、不同的金额字段。统一之后业务层只写一行。
## 五、同一个需求，两种写法
需求：「拉三个平台昨天的订单，算出各平台待发货数量和金额。」
**没有统一接口：**
```python
# 30 多行，而且每加一个平台就再加 10 行
result = {}
# 淘宝
t_orders, page = [], 1
while True:
d = taobao.call("taobao.trades.sold.get", {...})
rows = d["trades_sold_get_response"]["trades"]["trade"]
t_orders += rows
if len(rows) < 100:
break
page += 1
result["taobao"] = {
"pending": sum(1 for o in t_orders if o["status"] == "WAIT_SELLER_SEND_GOODS"),
"amount": sum(float(o["payment"]) for o in t_orders
if o["status"] == "WAIT_SELLER_SEND_GOODS"),
}
# 1688（字段名完全不同）
a_orders, page = [], 1
while True:
d = alibaba.call("com.alibaba.trade/get.soldList", {...})
rows = d["result"]["orderList"]
a_orders += rows
if len(rows) < 50:
break
page += 1
result["alibaba"] = {
"pending": sum(1 for o in a_orders if o["orderStatus"] == "waitsellersend"),
"amount": sum(float(o["sumPayment"] or 0) for o in a_orders
if o["orderStatus"] == "waitsellersend"),
}
# 京东（又是另一套）
...
```
**有统一接口：**
```python
summary = await api.order_summary("2026-10-22 00:00:00", "2026-10-23 00:00:00")
pending = {p: v["by_status"].get("PAID", 0) for p, v in summary.items()}
```
从 30 行到一个方法调用。而且**加第四个平台，业务代码一行都不用改。**
## 六、能力不对等：统一接口最硬的部分
这是统一接口真正难的地方，也是最容易埋雷的地方。
三个平台的能力根本不对等：
| 能力 | 淘宝 | 1688 | 京东 |
| --------- | --- | ----------- | --------- |
| SKU 级库存 | 支持 | 不支持（按起订量阶梯） | 部分接口支持 |
| 阶梯价 | 不支持 | 支持 | 不支持 |
| 订单按修改时间增量 | 支持 | 支持 | 部分接口只支持按天 |
| 物流订阅推送 | 支持 | 需单独申请 | 支持 |
| 商品批量查询 | 不支持 | 支持 | 支持 |
**最坑的做法是「假装都支持」**：调用方传 `sku_id`，1688 适配器忽略这个参数返回商品总量。业务代码以为拿到了 SKU 库存，实际上是错的，而且不报错。
所以 `get_stock` 里那句显式 `raise NotSupported` 才是正确做法：
```python
if sku_id:
raise NotSupported(self.platform, "按 SKU 查询库存（1688 无此概念）")
```
配套的用法是**启动时检查能力**：
```python
def check_capabilities(api: CommerceAPI, required: dict[str, bool]):
"""
启动时校验：业务需要的能力，各平台是否都支持。
不支持就提前暴露，而不是跑到线上才发现。
"""
problems = []
for platform, ad in api.adapters.items():
for cap_name, needed in required.items():
if needed and not getattr(ad.capability, cap_name, False):
problems.append(f"{platform} 不支持 {cap_name}")
if problems:
raise RuntimeError("能力校验失败：\n" + "\n".join(problems))
# 启动时调用
check_capabilities(api, {
"sku_level_stock": True, # 业务强依赖 SKU 库存
"order_increment": True,
})
```
这段会在启动时告诉你：「1688 不支持 sku_level_stock」。你可以立刻决定是降级、还是排除 1688，而不是等三天后客户投诉库存不准。
## 七、一套测试跑三个平台
统一接口的「统一」不能靠嘴说，得靠测试证明。用 pytest 参数化，同一套契约测试跑三个适配器：
```python
# tests/test_contract.py
import pytest
from decimal import Decimal
from core.adapter import NotSupported
@pytest.fixture
def adapters():
"""三个适配器实例，用测试账号"""
return {"taobao": ..., "alibaba": ..., "jd": ...}
@pytest.mark.parametrize("platform", ["taobao", "alibaba", "jd"])
@pytest.mark.asyncio
async def test_product_shape(adapters, platform, sample_ids):
"""契约一：商品必须返回统一结构"""
p = await adapters[platform].get_product(sample_ids[platform])
assert p.platform == platform
assert isinstance(p.title, str)
assert isinstance(p.price, Decimal)
assert p.price >= 0
assert isinstance(p.images, list)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
@pytest.mark.parametrize("platform", ["taobao", "alibaba", "jd"])
@pytest.mark.asyncio
async def test_order_status_normalized(adapters, platform):
"""契约二：状态必须是五个统一值之一"""
orders = await adapters[platform].list_orders(
"2026-10-15 00:00:00", "2026-10-23 00:00:00")
for o in orders:
assert o.status in {"PENDING", "PAID", "SHIPPED", "DONE", "CLOSED", "UNKNOWN"}
assert isinstance(o.pay_amount, Decimal)
@pytest.mark.asyncio
async def test_unsupported_explicit(adapters, sample_ids):
"""契约三：不支持的能力必须明确报错，不能静默降级"""
with pytest.raises(NotSupported):
await adapters["alibaba"].get_stock(
sample_ids["alibaba"], sku_id="any_sku")
```
第三条测试是整套设计里最重要的一条。它把「不能静默降级」这个约定固化成了测试，防止后来的人图省事改成返回 `0`。
## 八、收益复盘
同一套业务代码，服务三个平台：
| 指标 | 三套实现 | 统一接口层 |
| ------ | ----------- | -------------- |
| 首次对接 | 18 天 | 13 天（省 28%） |
| 一年维护 | 约 43 天 | 约 16 天（省 63%） |
| 业务层代码量 | 30 行 / 每个需求 | 3 行 / 每个需求 |
| 加第四个平台 | 约 5 天 | 约 1.5 天（省 70%） |
| 新人上手 | 读三套代码 | 读一个门面 + 三个薄适配器 |
综合下来，**一年周期省 80% 左右是站得住的**。但前提是：真的长期跑，且平台数量 ≥ 2。
只接一个平台，或者只做一次性的数据导出，这个抽象就是过度设计。工程决策永远看场景。
## 九、四个坑
1. **别在适配器里写业务判断。** 适配器只做翻译。一旦出现 `if 订单金额 > 100` 这种代码，说明业务逻辑漏进了适配器，将来三个平台会各写一份不一样的规则。
2. **能力声明要跟着平台文档更新。** 京东的接口能力经常变（同一个能力，不同接口支持度不同）。建议每个季度对照文档核一次 `Capability`，把它当配置维护，不是当常量。
3. **统一状态不要贪多。** 五个状态够用。给三个平台各定十个细分状态，最后业务代码里全是 `if status in [...]`，反而更乱。
4. **门面不要变成上帝类。** `CommerceAPI` 只放跨平台共性（重试、缓存、并发、能力检查）。具体业务（比如「降价 5% 就告警」）放在 service 层，别塞进门面。
## 十、最后
「一套接口打三平台」的价值，不在少写几行代码，在**让「平台差异」变成一个有边界、可枚举、可测试的东西**。
差异集中在适配器里，业务代码里就再也不该出现 `if platform == "taobao"`。一旦出现了，说明抽象漏了。
代码在上面，`BaseAdapter` + `Capability` + 契约测试这三件套可以直接搬到你的项目里。
如果你连适配器都不想写——签名、字段映射、状态归一、能力差异这些事，现成的三平台 API 接口服务已经做完了，输出就是统一格式，接进来只需要写业务层。省下的正是前面算的那 13 天加 16 天。
有在做的可以聊聊，你们的统一层是怎么处理能力不对等的，是抛错还是降级？
---
**标签**：`Python` `架构设计` `电商` `适配器模式` `API`未分类
从 3 小时到 3 分钟：三平台商品同步是怎么提速的（附 Python 源码）
万邦科技Lex 发表于2026-10-23 浏览15 评论0
# 从 3 小时到 3 分钟：三平台商品同步是怎么提速的（附 Python 源码）
> 60 倍提速不是靠某一个技巧，是四个优化叠加。每个贡献多少，本文算给你看。
## 一、一个真实的慢
某个多平台商家，3000 个商品，每天要同步一次三平台的商品数据（标题、价格、库存、状态）。
原来的脚本：串行逐个拉，每个商品调一次详情接口，拉完逐条写库。跑一次 3 小时。
运营的抱怨很直接：「我早上想看昨天的库存，脚本要到中午才跑完，看的是上午的数据。」
更糟的是，**没人说得清 3 小时都花在哪**。有人说接口慢，有人说服务器不行，有人说数据库写不动。都在猜。
## 二、第一步：先测量，别猜
性能优化的第一原则：**先测量，再动手。** 我在这上面吃过亏——曾经花一天优化数据库索引，后来 profile 一跑发现数据库只占 2% 的时间。
给脚本加了个简单的计时器，跑一次全量，结果如下：
| 环节 | 耗时 | 占比 |
| ---------------- | ------ | ---- |
| 调接口等响应（3000 次串行） | 172 分钟 | 96% |
| JSON 解析 + 字段映射 | 4 分钟 | 2% |
| 逐条写数据库 | 3 分钟 | 2% |
| 其他（重试、日志） | 1 分钟 | 0.5% |
结论非常清晰：**96% 的时间在等网络**。
这不是「接口慢」，是**串行等 3000 次**。每个请求平均 3.4 秒，串行乘以 3000，就是 172 分钟。
这个测量直接决定了优化顺序：先解决等待（并发），再减少请求数（增量），写库那 3 分钟排最后。
测量代码很简单，就几十行，但它决定了后面所有力气往哪使：
```python
# timer.py
import time
from collections import defaultdict
class StageTimer:
"""分阶段计时，跑完输出每阶段耗时和占比"""
def __init__(self):
self.spent = defaultdict(float)
self.counts = defaultdict(int)
self._t0 = None
self._current = None
def start(self, stage: str):
self._current = stage
self._t0 = time.perf_counter()
def stop(self):
if self._current and self._t0 is not None:
self.spent[self._current] += time.perf_counter() - self._t0
self.counts[self._current] += 1
self._current = None
def report(self) -> list[tuple[str, float, float]]:
total = sum(self.spent.values()) or 1.0
rows = sorted(self.spent.items(), key=lambda x: -x[1])
return [(name, sec, sec / total) for name, sec in rows]
def print(self):
print(f"\n{'环节':<24}{'次数':>8}{'耗时':>12}{'占比':>8}")
for name, sec, pct in self.report():
print(f"{name:<24}{self.counts[name]:>8}{sec:>10.1f}s{pct:>8.1%}")
```
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
## 三、优化一：并发——最直觉，但有天花板
96% 在等网络，那并发就是第一刀。把 3000 个串行请求改成 20 路并发：
```python
# concurrent.py
import asyncio
import time
class RateGate:
"""
匀速限流闸门：保证全进程 QPS 不超过上限。
比每个任务自己 sleep 更准，因为它是全局视角。
"""
def __init__(self, qps: float):
self.interval = 1.0 / qps
self._next_slot = time.monotonic()
self._lock = asyncio.Lock()
async def wait(self):
async with self._lock:
now = time.monotonic()
wait = self._next_slot - now
if wait > 0:
await asyncio.sleep(wait)
self._next_slot += self.interval
else:
self._next_slot = now + self.interval
async def fetch_all(ids: list[str], fetch_one, concurrency: int = 20,
qps: float = 10.0) -> list:
"""
并发拉取。两个旋钮：
- concurrency：同时在飞的请求数（受限于连接和服务端）
- qps：每秒发起多少个（受限于平台限流）
两个都要设，只设 concurrency 会被限流打回来
"""
gate = RateGate(qps)
sem = asyncio.Semaphore(concurrency)
results = [None] * len(ids)
async def one(idx: int, item_id: str):
async with sem:
await gate.wait()
results[idx] = await fetch_one(item_id)
await asyncio.gather(*[one(i, x) for i, x in enumerate(ids)])
return results
```
注意这段代码里有两个旋钮：`concurrency`（同时在飞的请求数）和 `qps`（每秒发起多少）。
**只设并发不设 QPS，是新手最常见的翻车点。** 20 路并发、每秒发出 20 个请求，直接撞平台限流，然后全部进入重试，反而更慢。
实测：3 小时 → 9 分钟。20 倍。
但天花板立刻出现了：平台限流 QPS 10，3000 个商品理论下限就是 300 秒 = 5 分钟。**并发再高也没用，限流是硬约束。**
所以第二刀必须换思路：不是「把请求发得更快」，而是「发更少的请求」。
## 四、优化二：少发请求——本文最值钱的洞察
重新看需求：3000 个商品每天同步。真的 3000 个都变了吗？
统计了一下：**每天真正变化的商品不到 10%**（改价、改库存、上下架）。90% 的请求在拉「没变的数据」。
那能不能先知道哪些变了，再只拉那些？
能。**商品列表接口一次能返回几十条记录的 `modified` 时间戳**——用 100 次列表请求（每次 30 条）就能拿到全部 3000 个商品的更新时间，替代 3000 次详情请求。
```python
# incremental.py
import json
from datetime import datetime
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def list_all_modified(client, shop_id: str) -> dict[str, str]:
"""
分页拉全量商品列表，只取 item_id -> modified 映射。
3000 个商品 = 100 次轻量请求，代替 3000 次重量请求
"""
mapping = {}
page = 1
while True:
rows = client.list_products(shop_id, page=page, page_size=30)
if not rows:
break
for r in rows:
mapping[str(r["item_id"])] = r["modified"]
if len(rows) < 30:
break
page += 1
return mapping
def pick_changed(current: dict[str, str],
last_sync: dict[str, str] | None) -> list[str]:
"""
和上次同步记录对比，只返回有变化的 item_id
新商品 + 修改过的商品 + 已删除的商品（单独处理）
"""
last = last_sync or {}
changed = [iid for iid, m in current.items() if last.get(iid) != m]
removed = [iid for iid in last if iid not in current]
if removed:
print(f"[sync] {len(removed)} 个商品已下架，需清理")
return changed
def save_sync_state(state_path: str, current: dict[str, str]):
with open(state_path, "w", encoding="utf-8") as f:
json.dump(current, f, ensure_ascii=False)
def load_sync_state(state_path: str) -> dict[str, str] | None:
try:
with open(state_path, encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
return None
```
账算一下：
- 原来：3000 次详情请求（每次约 3.4 秒等效负载）
- 现在：100 次列表请求 + 约 300 次详情请求 = 400 次请求
**请求数降到原来的 13%。** 这比任何并发技巧的收益都大。
这个思路的本质：**用便宜的请求替代昂贵的请求。** 列表接口一次返回 30 条的元数据，单价是详情接口的三十分之一。先便宜地筛一遍，再对筛出来的做贵的操作。
## 五、优化三 + 四：写库和连接
到这一步，网络等待只剩 40 秒左右，剩下的小头也该收一收。
**逐条写库 → 批量写。** SQLite 用 `executemany`，MySQL 用多值 INSERT：
```python
# bulk_write.py
import sqlite3
def bulk_upsert(conn: sqlite3.Connection, rows: list[tuple]):
"""
批量 upsert。3000 条逐条写 = 3000 次事务；
executemany 一批 = 1 次事务，差两个数量级
"""
conn.executemany("""
INSERT INTO products(platform, item_id, title, price, stock, synced_at)
VALUES(?,?,?,?,?,datetime('now'))
ON CONFLICT(platform, item_id) DO UPDATE SET
title=excluded.title, price=excluded.price,
stock=excluded.stock, synced_at=excluded.synced_at
""", rows)
conn.commit()
```
**每次新建连接 → 复用连接。** 这个最隐蔽：每次请求新建 TCP + TLS 握手要 200~~400ms，3000 次就是 10~~20 分钟。`httpx.AsyncClient` 复用连接后，握手只发生一次。
```python
# 复用连接：整个进程共享一个 client
# 错误：每次 fetch_one 里 with httpx.AsyncClient() as c: ...
# 正确：模块级或应用级创建一次
import httpx
client = httpx.AsyncClient(
timeout=10.0,
limits=httpx.Limits(max_connections=20, max_keepalive_connections=20),
)
```
顺带把限流做成自适应（AIMD），限流报错就减半 QPS，平稳运行则缓慢恢复：
```python
# adaptive.py
import random
class AdaptiveRate:
"""
自适应限速（AIMD）：
- 撞限流：QPS 减半（Multiplicative Decrease）
- 连续 50 次成功：QPS +10%（Additive Increase）
比写死一个保守值快得多，也比写死激进值安全
"""
def __init__(self, qps: float = 5.0, min_qps: float = 0.5, max_qps: float = 20.0):
self.qps = qps
self.min_qps = min_qps
self.max_qps = max_qps
self._ok_streak = 0
self._next_slot = 0.0
def on_success(self):
self._ok_streak += 1
if self._ok_streak >= 50:
self.qps = min(self.max_qps, self.qps * 1.1)
self._ok_streak = 0
def on_rate_limited(self):
self.qps = max(self.min_qps, self.qps / 2)
self._ok_streak = 0
@property
def interval(self) -> float:
return 1.0 / self.qps
async def wait(self):
"""带抖动的等待，避免多任务同刻发车"""
import asyncio, time
now = time.monotonic()
wait = max(0.0, self._next_slot - now)
if wait:
await asyncio.sleep(wait)
self._next_slot = max(now, self._next_slot) + self.interval * random.uniform(0.8, 1.2)
```
## 六、结果：60 倍是怎么来的
四个优化叠加，每一步的贡献都有数：
| 阶段 | 耗时 | 相比上一步 | 手段 |
| ------------------ | ---------- | -------- | ------------------------------- |
| 原始 | 180 分钟 | — | 串行 + 全量 + 逐条写 |
| + 并发 20 + QPS 限流 | 9 分钟 | **20×** | asyncio 并发，全局 QPS 闸门 |
| + 列表预筛增量 | 约 1.2 分钟 | **7.5×** | 100 次列表替代 2820 次详情 |
| + 批量写 + 连接复用 + 自适应 | **约 1 分钟** | 1.3× | executemany + keep-alive + AIMD |
两个观察值得记住：
1. **最大的收益来自「少做事」（7.5×）而不是「做得快」（20× 有硬顶）。** 并发受限于平台限流，少发请求不受任何限制。
2. **最后一个优化只贡献 1.3×，但它是「从 1 分钟到 40 秒」和「偶尔翻车」的区别。** 连接复用和自适应限速不影响平均值，影响的是稳定性。
## 七、完整主流程
把四个优化串起来：
```python
# sync_main.py
import asyncio
import sqlite3
from timer import StageTimer
from concurrent import fetch_all
from incremental import (list_all_modified, pick_changed,
load_sync_state, save_sync_state)
from bulk_write import bulk_upsert
from adaptive import AdaptiveRate
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
async def run_sync(clients: dict, shop_id: str,
state_path: str = "sync_state.json"):
t = StageTimer()
# 1. 增量预筛：100 次轻量请求拿全量 modified
t.start("列表预筛(3平台)")
current = {}
for platform, client in clients.items():
current[platform] = await list_all_modified(client, shop_id)
t.stop()
# 2. 对比上次状态，只保留有变化的
last = load_sync_state(state_path)
t.start("增量对比")
changed_map = {p: pick_changed(m, last and last.get(p))
for p, m in current.items()}
t.stop()
# 3. 并发拉详情（自适应限速）
t.start("详情并发拉取")
rows = []
for platform, client in clients.items():
ids = changed_map[platform]
if not ids:
continue
rate = AdaptiveRate(qps=5.0)
async def fetch_one(iid):
await rate.wait()
try:
r = await client.get_product_detail(iid)
rate.on_success()
return (platform, iid, r["title"], float(r["price"]),
r["stock"])
except Exception as e:
if "limit" in str(e).lower():
rate.on_rate_limited()
raise
results = await fetch_all(ids, fetch_one,
concurrency=20, qps=10.0)
rows.extend(r for r in results if r)
t.stop()
# 4. 批量写库
t.start("批量写库")
conn = sqlite3.connect("products.db")
bulk_upsert(conn, rows)
conn.close()
t.stop()
# 5. 保存同步状态，供下次增量对比
save_sync_state(state_path, current)
t.print()
return len(rows)
if __name__ == "__main__":
from client import PlatformClient
clients = {
"taobao": PlatformClient("taobao", "key", "secret"),
"alibaba": PlatformClient("alibaba", "key", "secret"),
"jd": PlatformClient("jd", "key", "secret"),
}
asyncio.run(run_sync(clients, "your_shop_id"))
```
## 八、四个坑
1. **并发数和 QPS 是两个旋钮，必须分开调。** 只调并发会被限流打回来进入重试风暴，比串行还慢。
2. **增量依赖 `modified` 字段，但不是所有平台都可靠。** 有的平台批量改价不更新 `modified`。建议每周跑一次全量对账，增量为主、全量兜底——和价格同步那篇的思路一致。
1. **`fetch_one` 里闭包引用循环变量是经典 bug。** 上面代码里 `for platform, client in clients.items()` 里面定义 `fetch_one`，每次循环重新定义，没有引用问题；但如果把函数定义提到循环外再绑定变量，就会全部引用最后一次的 `client`。测试时三个平台的数据混在一起，多半是这个问题。
2. **首次运行没有同步状态，会走全量。** 这是正确行为（第一次本来就该全量），但要提醒使用者：第一次跑慢是正常的，从第二次开始才是 3 分钟。
## 九、最后
回头看这 60 倍：
- 并发（20×）：把等待重叠起来，但受限于平台限流
- 增量（7.5×）：**用便宜的请求替代昂贵的请求**，收益最大
- 批量写 + 连接复用（1.3×）：平均值贡献小，稳定性贡献大
顺序也重要：**先测量再动手**。如果先花一天优化数据库（只占 2%），一天就白费了。
代码在上面，`RateGate`、`AdaptiveRate`、增量预筛这三个组件是通用的，可以拆到任何同步场景里用。
有在做的可以聊聊，你们的同步脚本跑多久，profile 过吗？
---
**标签**：`Python` `性能优化` `电商` `异步编程` `数据同步`