[特殊字符]《京东库存API（jd.stock.get / jd.stock.occupyStock）实时同步方案：防超卖与高并发处理》（附Python源码）

《京东库存APIjd.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 urlhttps://console.open.onebound.cn/console/?iLex # ───────────── 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_asciiFalse, separators(,, :)), access_token: token } p[sign] self._sign(p) r requests.post(self.GW, datap, timeout15) 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(fJOS [{err.get(code)}]: f{err.get(zh_desc) or err.get(en_desc)}) raise Exception(fJOS 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) # ───────────── 本地库存缓存 防超卖预占 ───────────── class LocalStockCache: 内存缓存示例生产换 Redis Hash key sku_id value {stock: int, reserved: int, ts: datetime} def __init__(self, ttl_sec300): 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_sec300) # ① 定时校正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高并发超卖多进程未原子预占用 RedisDECRWATCH/ 分布式锁全量翻页超日额度不必要只拉有变更SKUmodified筛或按分类分批沙箱无库存数据正常切生产五、面试/方案一句话京东库存 定时拉jingdong.ware.stock.get(skuIds)刷新本地缓存 → 下单时本地原子预占校验防超卖 → 京东仓订单额外调jd.stock.occupyStock预占无实时公开扣减API时以本地安全库存短周期校正兜底。需要我补Redis 原子预占 Lua 脚本​ 或APScheduler 定时库存校正 断点续跑​ 完整版吗