📦《京东库存API（jd.stock.get / jd.stock.occupyStock）实时同步方案：防超卖与高并发处理》（附Python源码）
直接说结论先：
京东可售库存快照用jingdong.stock.get（或新版jingdong.ware.stock.get）按wareId/skuId拉取，免费额度内零成本；严格防超卖推荐组合：本地缓存库存快照 + 下单前实时校验京东库存 + 京东占用库存接口(jd.stock.occupyStock)/减少库存回写（若使用京东仓WMS），中台侧以本地预留安全库存(MOQ/安全库存)做预警，超卖高风险SKU设本地预占。
一、京东库存相关接口速览
接口 | 用途 | 注意 |
|---|---|---|
jingdong.stock.get(老) / jingdong.ware.stock.get(新推荐) | 查询SKU可售库存/在途/冻结 | 需卖家AccessToken（查自己店铺商品） |
jingdong.stock.occupyStock | 预占库存（京东仓订单创建时） | 仅当用京东仓WMS且下推JD订单 |
jingdong.stock.releaseOccupiedStock | 释放预占 | 取消/缺货释放 |
jingdong.etms.warehouse.stock.get | 多仓库存查询 | 多仓商家 |
❌ 无"实时锁定扣减"公开API | 第三方ERP需本地预占+定时校正 | 用快照+安全库存兜底 |
⚠️ 京东不提供像淘宝"实时锁定库存"那样的公开增值扣减接口给外部ERP，通常做法是：
occupyStock预占 → 建京东订单 → 自动扣stock.get校正，设安全库存预警
二、Python：库存快照拉取 + 本地防超卖预占示例
# jd_stock_sync.py
"""
京东库存实时同步 + 本地防超卖预占示例
- 拉取SKU可售库存(jingdong.ware.stock.get 或 jingdong.stock.get)
- 内存/Redis 缓存快照
- 下单时本地预占检查 → 超卖拒绝 → 定时校正
依赖: requests (pip install requests)
"""
import hashlib
import json
import requests
import time
from datetime import datetime
from typing import Dict, List, Optional
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ───────────── JOS Client (内联最小版) ─────────────
class JdStockClient:
GW = "https://api.jd.com/routerjson"
def __init__(self, ak, ask):
self.ak, self.ask = ak, ask
def _sign(self, p: Dict) -> str:
filt = sorted((k, v) for k, v in p.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"{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')}]: "
f"{err.get('zh_desc') or err.get('en_desc')}")
raise Exception(f"JOS err: {d}")
return data
# ─── 查SKU库存（新版推荐 jingdong.ware.stock.get）───
def get_sku_stock(self, token: str, sku_ids: List[str]) -> Dict:
"""
sku_ids: ['100012345678', '100087654321'] ≤20个
返回 {skuId: stockNum}
"""
biz = {"skuIds": ",".join(sku_ids)}
# 部分老账号用 jingdong.stock.get → 参数 area 可选
resp = self._call("jingdong.ware.stock.get", biz, token)
# 返回结构: {stockGetResponse:{stockInfos:[{skuId,stockNum,areaId}]}}
infos = (resp.get("stockGetResponse") or {}
).get("stockInfos") or []
return {str(i.get("skuId")): int(i.get("stockNum") or 0)
for i in infos}
# ─── (可选) 预占库存 — 仅京东仓WMS ───
def occupy(self, token, sku_id, num, order_id):
return self._call("jingdong.stock.occupyStock", {
"skuId": sku_id, "num": num, "outerOrderId": order_id
}, token)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# ───────────── 本地库存缓存 + 防超卖预占 ─────────────
class LocalStockCache:
"""
内存缓存示例（生产换 Redis Hash）
key = sku_id value = {'stock': int, 'reserved': int, 'ts': datetime}
"""
def __init__(self, ttl_sec=300):
self.cache: Dict[str, Dict] = {}
self.ttl = ttl_sec
def update(self, sku_id: str, stock: int):
self.cache[sku_id] = {
"stock": stock,
"reserved": 0,
"ts": datetime.now()
}
def is_stale(self, sku_id: str) -> bool:
rec = self.cache.get(sku_id)
return not rec or (datetime.now() - rec["ts"]).total_seconds() > self.ttl
def available(self, sku_id: str) -> int:
rec = self.cache.get(sku_id)
if not rec:
return 0
return rec["stock"] - rec["reserved"]
def reserve(self, sku_id: str, qty: int) -> bool:
"""尝试预占，超卖返回False"""
rec = self.cache.get(sku_id)
if not rec:
return False
avail = rec["stock"] - rec["reserved"]
if avail < qty:
return False # ← 超卖拦截
rec["reserved"] += qty
return True
def release(self, sku_id: str, qty: int):
rec = self.cache.get(sku_id)
if rec:
rec["reserved"] = max(0, rec["reserved"] - qty)
# =========================================================
# 使用示例
# =========================================================
if __name__ == "__main__":
client = JdStockClient("YOUR_JD_APP_KEY", "YOUR_JD_APP_SECRET")
SELLER_TOKEN = "SELLER_ACCESS_TOKEN" # 卖家OAuth
SKU_IDS = ["100012345678", "100087654321"]
cache = LocalStockCache(ttl_sec=300)
# ① 定时校正（APScheduler每5分钟跑一次）
try:
stocks = client.get_sku_stock(SELLER_TOKEN, SKU_IDS)
for sid, stk in stocks.items():
cache.update(sid, stk)
print(f"✅ 库存同步 sku={sid} 可售={stk}")
except Exception as e:
print("❌ 库存拉取失败:", e)
# ② 下单预占检查
SKU = SKU_IDS[0]
QTY = 2
if cache.is_stale(SKU):
# 强制刷新单SKU
stk = client.get_sku_stock(SELLER_TOKEN, [SKU])
cache.update(SKU, stk.get(SKU, 0))
if cache.reserve(SKU, QTY):
print(f"✔ 预占成功 sku={SKU} x{QTY}，剩余可售={cache.available(SKU)}")
# → 继续创建内部销售单
else:
print(f"✘ 库存不足 sku={SKU} 申请{xQTY} 可用{cache.available(SKU)}")
# → 提示用户 / 标记缺货三、防超卖分层策略（推荐生产方案）
┌──────────────────────────────────────────────────────┐ │ 前端加车 / 下单页 │ │ ① 读 LocalStockCache.available() │ │ ② 若缓存过期 → 实时调 jd.stock.get 刷新 │ │ ③ 本地原子预占(reserve) → 失败回滚提示售罄 │ └──────────────────────┬───────────────────────────────┘ │ 每N分钟(2~5min) ▼ 后台定时全量/增量 jd.stock.get 更新 LocalStockCache(stock值) │ 若京东仓订单 → 调 jd.stock.occupyStock 自建仓 → 本地预占即够，出库后减本地库存
措施 | 作用 |
|---|---|
本地缓存+短TTL(2~5min) | 减少API调用，削峰 |
下单前实时校验（缓存过期刷一次） | 捕捉最近变化 |
本地预占原子操作 | 同进程/Redis DECR 防并发超卖 |
安全库存（MOQ×N） | 低于阈值标黄，不推此SKU做主推 |
京东仓占用接口 | 真正锁库存（仅限京东仓WMS推单） |
四、避坑清单
坑 | 现象 | 解决 |
|---|---|---|
查库存返回空 | 未传卖家 AccessToken / 应用无权限 | 订单类应用已申请， access_token是卖家OAuth |
stockNum异常大 | 部分虚拟/预售商品 | 结合 stockStatus判断（正常在售=1） |
高并发超卖 | 多进程未原子预占 | 用 Redis DECR+ WATCH/ 分布式锁 |
全量翻页超日额度 | 不必要 | 只拉有变更SKU（ modified筛）或按分类分批 |
沙箱无库存数据 | 正常 | 切生产 |
五、面试/方案一句话
京东库存 = 定时拉jingdong.ware.stock.get(skuIds)刷新本地缓存 → 下单时本地原子预占校验防超卖 → 京东仓订单额外调jd.stock.occupyStock预占；无实时公开扣减API时以本地安全库存+短周期校正兜底。
需要我补 Redis 原子预占 Lua 脚本 或 APScheduler 定时库存校正 + 断点续跑 完整版吗？
