多店铺京东订单统一拉取的核心在于：每个店铺独立 AccessToken + 增量时间窗拉取 jingdong.pop.order.search + 以 (tid, shop_id) 为唯一键做数据库幂等 + 分布式锁/Redis NX 防多节点并发重复处理 + 断点续跑防漏单。下面给你一套可直接落地的 Python 实战源码。
一、整体架构
┌────────────────────┐
│ 分布式调度器 │ APScheduler / Celery Beat
│ 遍历 shop_auth 表 │
└───────┬────────────┘
│ 每店铺独立 token
▼
┌────────────────────┐
│ JdOrderSyncPerShop │
│ 1. 获取分布式锁 │ Redis SET shop_id:sync NX EX
│ 2. 增量拉取订单 │ start_modified ~ end_modified
│ 3. 逐单查明细 │ jd.order.detail.get
│ 4. 幂等入库 │ INSERT ON CONFLICT(tid,shop_id) DO NOTHING
│ 5. 更新断点时间 │ last_sync_time
│ 6. 释放锁 │
└────────────────────┘
二、核心源码（Python · 含签名 / 分布式锁 / 幂等 / 断点）
jd_multi_shop_order_sync.py
"""
多店铺京东订单统一拉取实战
- JOS MD5签名（秒级timestamp）
- Redis分布式锁（防止多节点同时跑同一店铺）
- 增量同步 modified 时间窗
- 幂等入库 (tid, shop_id) UNIQUE
- 断点续跑（记录 last_sync_time）
依赖:
pip install requests redis apscheduler
"""
import hashlib
import json
import time
import datetime
import requests
import redis
from typing import Dict, List
================== 配置 ==================
APP_KEY = "YOUR_JD_APP_KEY"
APP_SECRET = "YOUR_JD_APP_SECRET"
REDIS_HOST = "localhost"
REDIS_PORT = 6379
REDIS_DB = 0
SYNC_INTERVAL_MIN = 30 # 增量窗口分钟数
LOCK_TTL_SEC = 300 # 分布式锁过期时间
==========================================
rds = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
---------- JOS Client ----------
class JdJosClient:
GW = "https://api.jd.com/routerjson"
def __init__(self, app_key, app_secret):
, self.ask = app_key, app_secret
def _sign(self, params: Dict) -> str:
filt = sorted(
(k, v) for k, v in params.items()
if k != "sign" and v is not None and str(v).strip() != ""
)
qs = "".join(f"{k}{v}" for k, v in filt)
return hashlib.md5(f"{self.ask}{qs}{self.ask}".encode()).hexdigest().upper()
def call(self, method, biz, token):
p = {
"app_key": self.ak,
"method": method,
"timestamp": str(int(time.time())), # 秒级！
"format": "json",
"v": "2.0",
"sign_method": "md5",
"360buy_param_json": json.dumps(biz, ensure_ascii=False,
separators=(',', ':')),
"access_token": token
}
p["sign"] = self._sign(p)
r = requests.post(self.GW, data=p, timeout=15)
r.raise_for_status()
d = r.json()
resp_key = method.replace(".", "_") + "_response"
if resp_key not in d:
for k in d:
if k.endswith("_response"):
resp_key = k
break
data = d.get(resp_key, d)
if "error_response" in str(data):
err = d.get(resp_key, {}).get("error_response") or d.get("error_response")
if err:
raise Exception(f"JOS[{err.get('code')}]: {err.get('zh_desc')}")
return data
def list_orders(self, token, start, end, page=1, page_size=100):
return self.call("jingdong.pop.order.search", {
"start_modified": start,
"end_modified": end,
"order_state": "WAIT_SELLER_STOCK_OUT,WAIT_BUYER_CONFIRM_GOODS,FINISHED",
"page": page,
"page_size": min(page_size, 100)
}, token)
def order_detail(self, token, order_id):
return self.call("jd.order.detail.get", {
"orderId": str(order_id),
"fields": "orderId,status,payment,buyer_nick,receiver_name,"
"receiver_mobile,orders.skuId,orders.outerSkuId,"
"orders.num,orders.price,created,modified"
}, token)
---------- 分布式锁 ----------
def acquire_lock(shop_id: str) -> str:
"""Redis SET NX EX，返回 identifier 或 None"""
identifier = str(time.time())
lock_key = f"jd_sync_lock:{shop_id}"
ok = rds.set(lock_key, identifier, nx=True, ex=LOCK_TTL_SEC)
return identifier if ok else None
def release_lock(shop_id: str, identifier: str):
lock_key = f"jd_sync_lock:{shop_id}"
# Lua等效：仅当值匹配才删，防误释放
with rds.pipeline() as pipe:
while True:
try:
pipe.watch(lock_key)
if pipe.get(lock_key) == identifier.encode():
pipe.multi()
pipe.delete(lock_key)
pipe.execute()
break
except redis.WatchError:
continue
---------- 断点 & 幂等（SQLite示例，生产换 MySQL/PG） ----------
def init_db(db="jd_erp.db"):
conn = import("sqlite3").connect(db)
conn.execute("""
CREATE TABLE IF NOT EXISTS erp_order(
tid TEXT,
shop_id TEXT,
status TEXT,
payment REAL,
buyer_nick TEXT,
created TEXT,
PRIMARY KEY(tid, shop_id)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS shop_sync_point(
shop_id TEXT PRIMARY KEY,
last_sync_time TEXT
)
""")
conn.commit()
return conn
def load_last_time(conn, shop_id):
cur = conn.execute("SELECT last_sync_time FROM shop_sync_point WHERE shop_id=?",
(shop_id,))
row = cur.fetchone()
return row[0] if row else None
def save_last_time(conn, shop_id, t):
conn.execute("""
INSERT INTO shop_sync_point(shop_id,last_sync_time)
VALUES(?,?) ON CONFLICT(shop_id) DO UPDATE SET last_sync_time=excluded.last_sync_time
""", (shop_id, t))
conn.commit()
def upsert_order(conn, shop_id, order):
conn.execute("""
INSERT INTO erp_order(tid,shop_id,status,payment,buyer_nick,created)
VALUES(?,?,?,?,?,?)
ON CONFLICT(tid,shop_id) DO NOTHING
""", (
str(order.get("orderId")),
shop_id,
order.get("status"),
float(order.get("payment") or 0),
order.get("buyer_nick", ""),
order.get("created", "")
))
conn.commit()
---------- 单店铺同步 ----------
def sync_one_shop(client: JdJosClient, conn, shop: Dict):
shop_id = shop["shop_id"]
token = shop["access_token"]
lock_id = acquire_lock(shop_id)
if not lock_id:
print(f"⚠️ 店铺[{shop_id}] 正在其他节点同步，跳过")
return
try:
now = datetime.datetime.now()
last = load_last_time(conn, shop_id)
if last:
start = last
else:
start = (now - datetime.timedelta(minutes=SYNC_INTERVAL_MIN * 2)
).strftime("%Y-%m-%d %H:%M:%S")
end = now.strftime("%Y-%m-%d %H:%M:%S")
print(f"▶ 店铺[{shop_id}] 增量 {start} ~ {end}")
page = 1
while True:
rsp = client.list_orders(token, start, end, page=page)
order_list = (rsp.get("popOrderSearch", {})
.get("orderSearch", {})
.get("orderInfoList", [])) or []
total = rsp.get("popOrderSearch", {}).get("orderSearch", {}
).get("orderTotal", 0)
for o in order_list:
tid = str(o.get("orderId"))
# 明细（按需，列表已有基础字段可先upsert）
detail = client.order_detail(token, tid)
order_data = detail.get("orderInfo", o)
upsert_order(conn, shop_id, order_data)
if len(order_list) < 100 or page * 100 >= total:
break
page += 1
time.sleep(0.15) # QPS保护
# 成功更新断点（用当前end，下次从此继续）
save_last_time(conn, shop_id, end)
print(f"✅ 店铺[{shop_id}] 同步完成，断点更新至 {end}")
except Exception as e:
print(f"❌ 店铺[{shop_id}] 同步异常: {e}")
finally:
release_lock(shop_id, lock_id)
---------- 多店铺入口 ----------
def sync_all_shops(shop_list: List[Dict]):
client = JdJosClient(APP_KEY, APP_SECRET)
conn = init_db()
for shop in shop_list:
try:
sync_one_shop(client, conn, shop)
except Exception as e:
print(f"❌ 顶层异常 shop={shop.get('shop_id')}: {e}")
===================== 示例 shop_list =====================
if name == "main":
SHOP_LIST = [
{"shop_id": "SHOP001", "access_token": "SELLER_TOKEN_A"},
{"shop_id": "SHOP002", "access_token": "SELLER_TOKEN_B"},
]
sync_all_shops(SHOP_LIST)
# ---- 常驻调度示例（取消注释）----
# from apscheduler.schedulers.blocking import BlockingScheduler
# sched = BlockingScheduler()
# sched.add_job(lambda: sync_all_shops(SHOP_LIST),
# 'interval', minutes=SYNC_INTERVAL_MIN, id='jd_multi_shop_sync')
# sched.start()
三、关键设计点说明
多店铺隔离
每个店铺一条 shop_auth 记录，存 shop_id / access_token / refresh_token，同步时按 shop_id 独立跑、独立锁、独立断点时间。分布式锁（Redis SET NX EX）
防止多节点 / 多进程同时跑同一店铺导致重复调用API、重复入库。幂等设计
本地订单表 PRIMARY KEY (tid, shop_id)，处理前数据库层 ON CONFLICT DO NOTHING，即使重试/重复推送也不重复建单。增量 + 断点续跑
记录 last_sync_time，每次用 start_modified=last_sync_time, end_modified=now 拉取，服务重启从中继续，避免漏单与超日额度。QPS保护
分页间 time.sleep(0.15) + 令牌桶（前文京东限流篇已给），遇 code=16 / ISP_FLOW_CONTROL 指数退避重试。对账兜底
每日凌晨全量比对一次最近 N 天订单，修正极端情况下漏掉的状态变更（回调丢失/网络抖动）。
四、避坑清单
坑 现象 解决
用毫秒时间戳 Invalid Timestamp JOS用秒级 int(time.time())
订单403 传买家token / 个人应用 企业认证 + 卖家OAuth AccessToken
多节点重复入库 无幂等/无锁 (tid,shop_id) 唯一索引 + Redis锁
重启后重复拉全量 无断点记录 持久化 last_sync_time
超日额度 limited-by-app-access-count 增量同步 + 断点，避免全量翻页
一句话总结：多店铺京东订单同步 = 每店铺独立 Seller Token → 增量 modified 时间窗 → Redis NX 锁防并发 → (tid,shop_id) 幂等入库 → 记录 last_sync_time 断点续跑 → 日对账兜底，签名/QPS 同单店铺，仅调度与隔离维度不同。
