《京东订单API jingdong.pop.order.search 落地方案：鼎内调用与QPS配额管理》（附 Python 源码）
一、先拍结论
jingdong.pop.order.search是 POP 商家订单检索接口，不是订单真相源（真相是jd.order.detail.get/ 订单消息）。它的生产定位是：“按状态+时间窗捞订单列表”，再拿orderId去拉详情。官方已经明说：
创建时间支持查近 2 年，按更新时间只支持近 3 个月
开始/结束时间不得相差超过 1 个月
“建议通过时间滚动的方式拉取订单，页数要尽量少”
云鼎内调用是官方标注的调用方式之一
所以京东订单同步的成熟写法不是“翻页翻到天荒地老”，而是：
时间窗滚动（30天/窗） → pop.order.search 捞 orderId → jd.order.detail.get 拉真相 → (order_id, order_state, operate_time) 做幂等 → JMQ/订单消息做触发器，search 做兜底
二、接口边界（别和这些搞混）
接口 | 作用 | 备注 |
|---|---|---|
jingdong.pop.order.search | POP 订单列表检索 | 创建时间近2年 / 更新时间近3月 / 时间窗≤1月 |
jingdong.pop.order.fbp.search | FBP 订单检索 | 京东仓配履约订单 |
jd.order.detail.get | 单订单详情 | 子单/优惠/发票/收货 |
JMQ 订单消息 | 变更推送 | 触发器，不是真相 |
jingdong.pop.order.print.data.get | 发货/打单数据 | 打单用 |
jingdong.etms.waybill.* | 电子面单/轨迹 | 履约用 |
POP 和 VOP（企业采购）权限完全隔离，别用同一个 AppKey 想通吃。
三、时间窗与分页：官方给的“反模式”
官方文档原话要点：
start_date/end_date差值 不得超过 1 个月WAIT_SELLER_STOCK_OUT时起止可空，默认前一个月其他状态
start_date必传page_size最大 100“建议您之前频繁的翻页，页数较大，请尽快调整为时间滚动方式”
❌ 错误写法
page=1..5000，固定最近30天，order_state=ALL
问题：
大商家一窗几万单 → 翻页深 → 超时/限流
订单在窗口内被修改 → 翻页过程中顺序漂移 → 漏单
状态 ALL + 深翻页 = 京东最喜欢限你流
✅ 正确写法
按 order_state 分桶 每个状态：时间窗 30 天，sortType=0 升序，page 不超 50 窗滚动：end = now - 30s，start = end - 30d 每次同步记录 (state, last_operate_time) 下次从 last_operate_time 继续
四、鼎内调用是什么
京东有“云鼎/鼎内”概念：应用/服务在京东云鼎环境内调用，某些订单/数据/合规链路更稳，官方接口文档也会标“云鼎内调用”。
生产含义：
鼎内：延迟低、订单/隐私字段合规链路更完整、大促抖动小
鼎外（公网 IDC）：能调，但隐私字段/大促稳定性/审计不如鼎内
订单类接口必须卖家 OAuth access_token，不管鼎内鼎外
经验口径（非官方保证）：京东商家 JOS 订单类接口，鼎内通常比鼎外更适合做“持续同步 + 大促兜底”；但“鼎内”不是免限流金牌，QPS 配额照样按应用/权限包来。
五、QPS 与配额：社区经验口径
官方不把“所有应用的 QPS”写死成一刀切，社区实测大致是：
应用类型 | 订单类经验 QPS | 说明 |
|---|---|---|
个人应用 | 1~2/s | 订单类基本别上生产 |
企业实名（基础） | 5~10/s | pop.order.search 常见档 |
买资源包 | 20 / 50 / 100/s | 需平台审批/购包 |
鼎内企业应用 | 视权限包 | 不是无限，别睡大觉 |
费用口径（社区经验）：
商家 JOS 有日免费额度
pop.order.search超量按量计费鼎内/云内单价低于公网（类似淘宝聚石塔内外价差逻辑）
联盟 API（
jd.union.open.*）是 CPS 选品体系，和商家订单同步是两回事
六、Python：京东 JOS 基础 Client（秒级时间！）
# jd/jos_client.py
import hashlib
import time
import requests
class JdJosClient:
GATEWAY = "https://api.jd.com/routerjson"
def __init__(self, app_key, app_secret, access_token=None, timeout=5):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
self.timeout = timeout
def _sign(self, params: dict) -> str:
flat = {k: v for k, v in params.items() if v is not None and k != "sign"}
qs = "".join(f"{k}{v}" for k, v in sorted(flat.items()))
raw = f"{self.app_secret}{qs}{self.app_secret}"
return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def execute(self, method: str, biz: dict):
params = {
"app_key": self.app_key,
"method": method,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), # 京东用秒级/字符串
"format": "json",
"v": "2.0",
"sign_method": "md5",
}
if self.access_token:
params["access_token"] = self.access_token
params.update(biz)
params["sign"] = self._sign(params)
resp = requests.post(self.GATEWAY, data=params, timeout=self.timeout)
return resp.json()京东坑：把 1688 的毫秒时间戳带过来 →timestamp invalid。京东要yyyy-MM-dd HH:mm:ss，允许误差 10 分钟。
七、Python：时间滚动订单检索器
# jd/pop_order_scanner.py
from datetime import datetime, timedelta
from ratelimit.token_bucket import TokenBucket
class PopOrderScanner:
MAX_WINDOW = timedelta(days=30)
SLIDE_BACK = timedelta(seconds=30) # 防修改时间漂移漏单
def __init__(self, client: JdJosClient, bucket: TokenBucket):
self.client = client
self.bucket = bucket
def scan_state(self, order_state: str, start: datetime, end: datetime):
while start < end:
window_end = min(start + self.MAX_WINDOW, end)
yield from self._scan_window(order_state, start, window_end)
# 下一窗往前滚，并回退 30s 防丢
start = window_end - self.SLIDE_BACK
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def _scan_window(self, order_state, start, end, page_size=100):
page = 1
while True:
self.bucket.acquire()
resp = self.client.execute("jingdong.pop.order.search", {
"start_date": start.strftime("%Y-%m-%d %H:%M:%S"),
"end_date": end.strftime("%Y-%m-%d %H:%M:%S"),
"order_state": order_state,
"dateType": 0, # 0=按创建时间；1=按修改时间
"sortType": 0, # 0=升序，滚动拉取稳
"page": str(page),
"page_size": str(page_size),
"optional_fields": (
"orderId,orderState,operateTime,orderStartTime,"
"itemInfoList,payType,orderSellerPrice,consigneeInfo"
),
})
node = resp.get("jingdong_pop_order_search_responce") or {}
result = node.get("searchorderinfo_result") or {}
api_result = result.get("apiResult") or {}
if api_result.get("success") == "false":
raise JdApiError(api_result.get("chineseErrCode"), api_result.get("numberCode"))
orders = result.get("orderInfoList") or []
if not orders:
break
for o in orders:
yield o
# 官方建议：别深翻页
if len(orders) < page_size or page >= 50:
break
page += 1状态分桶示例：
STATES = [ "NOT_PAY", "WAIT_SELLER_STOCK_OUT", "WAIT_GOODS_RECEIVE_CONFIRM", "FINISHED_L", "TRADE_CANCELED", "LOCKED", ]
八、QPS 配额管理器（单实例 + Redis 分布式）
单实例令牌桶
# ratelimit/token_bucket.py import time # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex class TokenBucket: def __init__(self, rate: float, capacity: float = None, jitter: float = 0.05): self.rate = rate self.capacity = capacity or rate self.tokens = self.capacity self.last = time.monotonic() self.jitter = jitter def acquire(self, n: int = 1): while True: now = time.monotonic() self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate) self.last = now if self.tokens >= n: self.tokens -= n return sleep = (n - self.tokens) / self.rate + self.jitter time.sleep(sleep)
多实例分布式限流
# ratelimit/redis_sliding.py
import time
class RedisQpsLimiter:
def __init__(self, redis, key: str, qps: int):
self.r = redis
self.key = key
self.qps = qps
def allow(self) -> bool:
now = time.time()
self.r.zremrangebyscore(self.key, 0, now - 1)
self.r.zadd(self.key, {str(now): now})
self.r.expire(self.key, 2)
return self.r.zcard(self.key) <= self.qps调用侧：
if not redis_limiter.allow(): time.sleep(0.1) continue bucket.acquire()
规则：
企业 10 QPS → 客户端设 8
鼎内企业 20 QPS → 客户端设 16
多店铺：按
shop_id分桶，别全店共享一个令牌桶列表 8 QPS + 详情 5 QPS 分开桶，别互相挤死
九、限流/错误分类（京东特有）
class JdApiError(Exception):
def __init__(self, msg, code=None):
super().__init__(msg)
self.code = code
def classify(resp):
node = resp.get("jingdong_pop_order_search_responce", {})
api = (node.get("searchorderinfo_result") or {}).get("apiResult") or {}
if api.get("success") == "false":
msg = api.get("chineseErrCode", "")
code = api.get("numberCode", "")
# 限额/限流
if "accesscontrol" in msg or "limited" in msg or str(code) == "16":
raise RetriableError(f"jd limit: {msg}")
# 参数/权限
if "permission" in msg or "validate" in msg:
raise FatalError(f"jd param/perm: {msg}")
raise RetriableError(f"jd api err: {msg}")
# HTTP 层
# code=16 / 429 / ISP_FLOW_CONTROL_LIMIT → 退避重试策略：
code=16/ 429 / flow control → 指数退避（10s, 30s, 60s）token 过期 → 刷 access_token 再试 1 次
参数错 / 无权限 / 店铺未授权 → 不重试，告警
网络超时 → 重试，但别把“超时”当成“没数据”
十、幂等与状态机
pop.order.search 只给列表，别直接写业务：def upsert_from_search(order):
order_id = str(order["orderId"])
state = order["orderState"]
op_time = parse_time(order.get("operateTime") or order.get("orderStartTime"))
cur = order_repo.get(order_id)
version = int(op_time.timestamp() * 1000)
if cur and version < cur.version:
return # 旧窗数据晚到
# 只存“列表级快照”，详情再去 jd.order.detail.get 拉
order_repo.upsert({
"order_id": order_id,
"order_state": state,
"operate_time": op_time,
"version": version,
"need_detail": True,
})详情拉取：
def fill_detail(order_id):
resp = client.execute("jd.order.detail.get", {"order_id": order_id})
trade = extract_trade(resp)
order_repo.update_detail(trade)十一、消息 + 搜索 双通道（生产必做）
JMQ 订单消息 └─> 入队 order_change(order_id) 定时器（每 60s） └─> pop.order.search 按状态滚动 └─> 补消息漏掉的 order_id 两者都只干一件事： mark need_detail = True Worker： jd.order.detail.get → 写 order_snapshot → 跑状态机
铁律：
消息：快返回、msgId 去重、异步处理
搜索：兜底、时间滚动、别深翻页
详情：真相源
(order_id, order_state, operate_time)：幂等键
十二、大促/店铺多时的配置
JD_POP_CONFIG = {
"bucket_qps": 8, # 企业 10 留余量
"detail_qps": 5,
"window_days": 30,
"slide_back_sec": 30,
"max_page": 50,
"states": STATES,
"per_shop_bucket": True,
"use_jmq": True,
"search_fallback_interval": 60, # 秒
"dingnei_preferred": True, # 鼎内网关/部署优先
}鼎内部署时：
订单同步服务放鼎内容器
Redis/DB 也走鼎内内网
公网只暴露“内部调度 API”，不暴露 JOS 调用层
大促前压测的是 detail + 发货 + 面单，不是 search 翻页
十三、落地检查表
[ ] 时间戳用
yyyy-MM-dd HH:mm:ss，别用毫秒[ ]
pop.order.search只捞列表，不写业务[ ] 时间窗 ≤ 30 天，状态分桶，回退 30s
[ ] page 不超 50，不深翻页
[ ] 列表 QPS / 详情 QPS / 面单 QPS 分桶
[ ] 多实例用 Redis 滑动窗口，不用单机 time.sleep 假装限流
[ ] 限流退避、token 过期刷一次、权限错告警
[ ] JMQ 做触发器，search 做兜底，detail 做真相
[ ]
(order_id, order_state, operate_time)幂等[ ] POP / FBP / VOP 应用分开，不串 AppKey
[ ] 鼎内优先，但鼎内也不假设“无限 QPS”
十四、一句话收口
jingdong.pop.order.search不是“订单同步主接口”，是“订单 ID 捕捞网”。生产里把它当主角 → 深翻页、限流、漏单、大促崩；把它当配角 → 时间滚动捞 ID、消息做闹钟、detail 做法官、QPS 桶做安全带。鼎内调用解决的是合规和稳定性，不解决“你不会限流”的问题。
