从0搭一个淘宝京东1688拼多多抖店五平台聚合中台核心不是“把五个SDK凑一起”而是用前几篇拆出的计费/限流/入塔规则倒推架构订单能推不拉、按平台着色部署、统一DTO收口、令牌桶按Key隔离、配额/余额守卫编进Client。下面给一套可直跑的轻量中台骨架Python单进程可启生产换Redis/Celery/Kafka即可。一、中台分层倒推出来的形态统一调度入口 Scheduler │ ├─ 平台Adapter层每平台一个Client签名/网关/Token刷新隔离 │ TaobaoAdapter(聚石塔内) JdAdapter Ali1688Adapter │ PddAdapter(云内余额守卫) DyAdapter(云内) │ ├─ 统一模型层 DTOStandardOrder / StandardSku / StandardStock │ ├─ 限流守卫层每AppKey独立令牌桶 日配额 拼多多余额熔断 │ ├─ 同步策略层推送消费为主 增量modified兜底 失败死信补偿 │ └─ 存储层PostgreSQL业务表 Redis幂等/计数/令牌桶设计铁律来自前五篇淘宝/抖店/拼多多必须云内否则×10倍或禁调订单DSS/Webhook/订单同步服务为主API增量仅兜底1688批发别硬轮询高级库存爆款走高级包或Webhook京东联盟Key与商家Key物理隔离拼多多欠费硬切断本地计数器兜底余额。二、统一DTO先把五家订单归一# dto.py from dataclasses import dataclass, field from enum import Enum from datetime import datetime class StdOrderStatus(str, Enum): CREATED CREATED PAID PAID SHIPPED SHIPPED SIGNED SIGNED REFUNDING REFUNDING CLOSED CLOSED dataclass class StandardOrder: channel: str # taobao/jd/ali1688/pdd/douyin shop_id: str order_id: str # 平台原始订单号 idempotency_key: str # channelorder_id status: StdOrderStatus StdOrderStatus.CREATED pay_amount: float 0.0 post_fee: float 0.0 item_count: int 0 buyer_remark: str created_at: datetime None modified_at: datetime None raw: dict field(default_factorydict) # 原始报文留存溯源 def __post_init__(self): if not self.idempotency_key: self.idempotency_key f{self.channel}:{self.shop_id}:{self.order_id}状态映射表各Adapter转换时查这张表STATUS_MAP { taobao: {WAIT_BUYER_PAY:CREATED,TRADE_PAID:PAID, WAIT_SELLER_SEND_GOODS:PAID,TRADE_BUYER_SIGNED:SIGNED, TRADE_CLOSED:CLOSED}, jd: {10:PAID,20:PAID,30:SHIPPED,40:SIGNED,60:CLOSED}, pdd: {0:CREATED,1:PAID,2:SHIPPED,3:SIGNED,5:REFUNDING}, douyin: {1:CREATED,2:PAID,3:SHIPPED,4:SIGNED,5:CLOSED}, ali1688: {waitbuyerpay:CREATED,waitsellersend:PAID, waitbuyerreceive:SHIPPED,confirm_send:SIGNED,cancel:CLOSED}, }三、按Key隔离的令牌桶 配额守卫核心# guard.py import time, hashlib, json, requests from datetime import datetime from threading import Lock class KeyRateGuard: 每个AppKey独立令牌桶限速 日调用计数 拼多多余额熔断 def __init__(self, platform, app_key, qps, daily_free, in_cloudTrue): self.platform platform self.app_key app_key self.qps qps self.tokens qps self.ts time.monotonic() self.lk Lock() self.day datetime.now().date() self.today_calls 0 self.daily_free daily_free self.in_cloud in_cloud self.pdd_balance None # 拼多多外部注入 def _roll_day(self): if datetime.now().date() ! self.day: with self.lk: self.day datetime.now().date() self.today_calls 0 def acquire(self, is_valueFalse): self._roll_day() # 1. 增值接口云外禁调 if is_value and not self.in_cloud and self.platform in (taobao,pdd,douyin): 朋 raise PermissionError(f{self.platform} 增值接口必须云内) # 2. 日免额80%预警100%熔断非核心 if self.today_calls self.daily_free: raise RuntimeError(f{self.app_key} 日免额{self.daily_free}耗尽停调防扣费) elif self.today_calls int(self.daily_free*0.8): print(f⚠️ {self.app_key} 达免额80%切纯增量) # 3. 拼多多余额守卫 if self.platformpdd and self.pdd_balance is not None: unit 0.01/100 if self.in_cloud else 0.10/100 if self.pdd_balance (self.today_calls1)*unit*3: raise RuntimeError(pdd 余额3天预估熔断) # 4. 令牌桶 with self.lk: now time.monotonic() self.tokens min(self.qps, self.tokens (now-self.ts)*self.qps) self.ts now if self.tokens 1: time.sleep((1-self.tokens)/self.qps 0.005) self.tokens 0 else: self.tokens - 1 self.today_calls 1四、五平台Adapter统一接口签名各异# adapters.py from abc import ABC, abstractmethod import hashlib, time, json, requests from dto import StandardOrder, STATUS_MAP class BaseAdapter(ABC): def __init__(self, guard: KeyRateGuard, app_key, app_secret): self.g guard self.ak app_key self.ask app_secret abstractmethod def pull_increment_orders(self, shop_id, token, start_mod, end_mod, page1) - list[StandardOrder]: ... def _sign_top_like(self, params): f sorted((k,v) for k,v in params.items() if k!sign and v is not None and str(v)!) qs .join(f{k}{v} for k,v in f) return hashlib.md5(f{self.ask}{qs}{self.ask}.encode()).hexdigest().upper() def _safe_req(self, url, params, is_valueFalse, max_retry4): self.g.acquire(is_value) params[sign] self._sign_top_like(params) for att in range(max_retry): try: r requests.post(url, dataparams, timeout15) d r.json() if error_response in d or errorResponse in d: blob json.dumps(d) if any(k in blob for k in (FLOW_CONTROL,limited-by,50001,no permission)): time.sleep(min(2**att,8)); continue raise Exception(blob) return d except requests.RequestException: time.sleep(2**att); continue raise RuntimeError(retry exhausted) class TaobaoAdapter(BaseAdapter): GW https://gw.api.taobao.com/router/rest def pull_increment_orders(self, shop_id, token, start_mod, end_mod, page1): biz {start_modified:start_mod,end_modified:end_mod, page_no:page,page_size:50,fields:tid,status,payment,post_fee,modified} p {method:taobao.trades.sold.increment.get,app_key:self.ak, timestamp:str(int(time.time()*1000)),format:json,v:2.0, sign_method:md5,access_token:token} p.update(biz) d self._safe_req(self.GW, p) out[] for t in d.get(trades_sold_increment_get_response,{}).get(trades,{}).get(trade,[]): out.append(StandardOrder( channeltaobao, shop_idshop_id, order_idstr(t[tid]), statusSTATUS_MAP[taobao].get(t[status],CREATED), pay_amountfloat(t.get(payment,0)), post_feefloat(t.get(post_fee,0)), modified_att.get(modified), rawt)) return out class PddAdapter(BaseAdapter): GW https://gw-api.pinduoduo.com/api/router def pull_increment_orders(self, shop_id, token, start_mod, end_mod, page1): p {client_id:self.ak,method:pdd.order.number.list.increment.get, timestamp:str(int(time.time())),data_type:JSON,v:V1.0, start_updated_at:int(start_mod),end_updated_at:int(end_mod), page:page,page_size:50,access_token:token} d self._safe_req(self.GW, p) out[] for o in d.get(order_number_list_increment_get_response,{}).get(order_list,[]): out.append(StandardOrder( channelpdd, shop_idshop_id, order_ido[order_sn], statusSTATUS_MAP[pdd].get(str(o[order_status]),CREATED), pay_amountfloat(o.get(pay_amount,0)), modified_ato.get(updated_at), rawo)) return out # JdAdapter / Ali1688Adapter / DyAdapter 同构略方法名一致签名换秒级/毫秒、method命名不同生产里把TaobaoAdapter/PddAdapter/JdAdapter/Ali1688Adapter/DyAdapter都实现同一抽象Scheduler不感知平台。五、统一调度器增量时间窗 多店轮转# scheduler.py import time from datetime import datetime, timedelta from adapters import TaobaoAdapter, PddAdapter from guard import KeyRateGuard class ShopBinding: def __init__(self, channel, shop_id, adapter, app_key, token, qps, daily_free, in_cloudTrue): self.channel channel self.shop_id shop_id self.adapter adapter self.token token self.guard KeyRateGuard(channel, app_key, qps, daily_free, in_cloud) # 注册中心实际从DB载 SHOPS [ ShopBinding(taobao,shopA,TaobaoAdapter(KeyRateGuard(taobao,AK_TB,8,80000), AK_TB,AS_TB), TB_TOKEN, 8, 80000, in_cloudTrue), ShopBinding(pdd,shopB,PddAdapter(KeyRateGuard(pdd,AK_PDD,8,50000), AK_PDD,AS_PDD), PDD_TOKEN, 8, 50000, in_cloudTrue), ] def sync_loop(): while True: end datetime.now() start end - timedelta(minutes5) # 5分钟增量窗 for sb in SHOPS: try: orders sb.adapter.pull_increment_orders( sb.shop_id, sb.token, start.strftime(%Y-%m-%d %H:%M:%S), end.strftime(%Y-%m-%d %H:%M:%S)) for o in orders: # 1. Redis幂等key存在则跳 # 2. 写PG standard_orderupsert by idempotency_key # 3. 发Kafka事件 order.updated print(f✔ {o.channel}/{o.shop_id}/{o.order_id} - {o.status}) except (RuntimeError, PermissionError) as e: print(f⚠️ {sb.channel}/{sb.shop_id} 守卫拦截: {e}) except Exception as e: print(f❌ {sb.channel}/{sb.shop_id} 异常: {e}) time.sleep(60) # 主控节拍1分钟内部增量5分钟窗 if __name__ __main__: sync_loop()关键点主控1分钟心跳拉取窗5分钟重叠防漏平台modified有秒级延迟每店独立Guard店铺A限流不影响店铺B守卫抛错不进DB只告警避免把限流当业务异常处理。六、推送为主的可插拔扩展点上面是“增量轮询兜底”版生产建议把各平台推送接进来淘宝聚石塔DSS订单推送 → 消费RDS Binlog/推送服务省API费拼多多订单同步服务多多云DB推送替代order.list.get抖店/1688消息订阅Webhook → MQ消费京东宙斯能力中心数据推送云鼎。调度器里加一个PushConsumer把消息转成StandardOrder走同一套幂等写轮询只作“每30分钟全量校对”的补偿任务。七、从0到1落地顺序避坑路径资质先行按前文认证表淘宝/抖店/拼多多订单必须企业自研应用1688高级库存买包京东商家JOS联盟隔离部署着色淘宝→聚石塔ECS抖店→抖店云拼多多→拼多多云1688/京东→同主体阿里云/京东云VPC先接推送每家开通订单推送/同步服务写StandardOrder落库再补轮询增量modified每5分钟兜底Guard卡80%免额商品/库存后接1688批发用高级包Webhook淘宝库存用skus.quantity.update回写别反向硬拉监控面板每AppKey日调用/剩余免额/拼多多余额/令牌桶等待长度 → 企微告警。这套骨架把“五家收费模型”编译进了代码云内强制校验、免额熔断、拼多多余额守卫、按Key令牌桶、统一DTO收口、增量重叠防漏。它不是最重的无Kafka/Celery但把多平台中台最易烂尾的“计费-限流-幂等”三件事在第一次启动时就焊死了。要不要我接着把PushConsumer淘宝DSS/拼多多同步服务/抖店Webhook 和PostgreSQL upsert Redis幂等键 的落地代码补完整让这套中台从“轮询骨架”升级成“推拉一体可上大促”的版本