简介这份仓库管理系统源码面向中小企业信息化负责人、物流仓储方向的开发者与学习者源自多年ERP项目实施经验将商业级WMS功能剥离精简而成聚焦小型物流仓储供应链流程支持跨平台、一处编码多处使用适合作为二次开发模板或教学范例。压缩包共445个文件约1.69MB以198个C#后端服务代码、105个Vue前端页面、86个TypeScript脚本为主体另含项目工程文件、配置与部署脚本、数据库文件及少量图片与说明文档覆盖入库、出库、库存、调拨等核心业务模块目录结构清晰便于按模块检索与复用。目前已有654人学习下载。读者可从中获得一套可直接运行的完整系统骨架理解WMS与ERP的功能边界划分参考服务层与前端页面的组织方式并借助现成配置快速搭建本地环境为自研仓储系统或课程设计提供可落地的实现思路。1. 从一张 Excel 台账说起简易完整的仓库管理系统到底该长什么样很多团队第一次做仓库管理都是从一张 Excel 台账开始的。入库加一行出库减一行月底对不上就翻聊天记录。等到 SKU 过千、多人同时改表、退货和调拨混在一起这张表就开始崩库存数对不上、谁改的查不到、超卖和积压同时出现。所谓「简易完整的仓库管理系统」不是把 ERP 那套几十个模块全搬过来而是用最小的一套数据模型和操作流程把「货在哪、有多少、怎么动的、谁动的」这四件事讲清楚并且能自己部署、自己改。它适合三类人一是中小电商或门店SKU 在几百到几万之间用不起也养不起重型系统二是做内部工具的后端或全栈工程师需要一套能跑通、能扩展的库存底座三是想理解 WMS 本质的学生和转行者需要一个能读完全部代码的完整样本。核心诉求就一句话进销存数据一致、操作可追溯、部署不折腾。下面按「数据模型怎么立 → 接口怎么写 → 并发怎么防 → 坑在哪 → 怎么验证」的顺序把一套能落地的方案讲透。2. 数据模型与表结构库存账不能只存一个数字2.1 为什么「库存数量」字段是万恶之源新手最容易犯的错是在商品表上直接放一个stock字段入库stock stock n出库stock stock - n。单机单人没问题一旦并发或需要追溯立刻翻车你只知道现在剩多少不知道这 100 件是怎么变成 80 件的退货、盘盈盘亏、调拨全糊在一起对账时就是一笔糊涂账。正确做法是引入「库存流水inventory transaction」作为唯一事实来源商品表上的库存数只是流水汇总出来的缓存。每一次库存变动都写一条流水记录变动类型、数量、方向、关联单据、操作人、时间。这样任何时刻的库存都能通过流水重算出来出问题能定位到具体哪一笔。这是 WMS 和「Excel 记账」最本质的区别也是后面所有功能的地基。常见做法是四张核心表商品表sku、仓库表warehouse、库存流水表stock_txn、库存快照表stock_snapshot可选用于加速查询。下面给出可直接建表的 SQL字段命名和类型都按 MySQL 8 的习惯来。-- 商品表只存商品静态信息不放实时库存 CREATE TABLE sku ( id BIGINT PRIMARY KEY AUTO_INCREMENT, sku_code VARCHAR(64) NOT NULL UNIQUE COMMENT 商品编码业务唯一键, name VARCHAR(128) NOT NULL, unit VARCHAR(16) DEFAULT 件, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 库存流水表唯一事实来源只增不改 CREATE TABLE stock_txn ( id BIGINT PRIMARY KEY AUTO_INCREMENT, sku_id BIGINT NOT NULL, warehouse_id BIGINT NOT NULL, txn_type TINYINT NOT NULL COMMENT 1入库 2出库 3调拨入 4调拨出 5盘盈 6盘亏, qty INT NOT NULL COMMENT 正数方向由 txn_type 决定, ref_no VARCHAR(64) COMMENT 关联单据号用于追溯, operator VARCHAR(64) NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, KEY idx_sku_wh (sku_id, warehouse_id, created_at) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 库存快照表按 SKU仓库 汇总读多写少时加速 CREATE TABLE stock_snapshot ( sku_id BIGINT NOT NULL, warehouse_id BIGINT NOT NULL, qty INT NOT NULL DEFAULT 0, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (sku_id, warehouse_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;逻辑说明stock_txn只插入不更新天然可审计txn_type用整数枚举而不是字符串索引更小、比较更快。参数上qty统一存正数方向交给txn_type判断避免正负号在业务代码里到处判断导致符号错误。idx_sku_wh这个联合索引是给「查某商品在某仓库的流水」用的顺序不能反否则范围查询用不上。提示如果 SKU 量不大几万以内可以不加stock_snapshot直接对stock_txn做SUM聚合逻辑更简单、更不容易出现快照和流水不一致。快照是性能优化不是必需品别一上来就引入。2.2 库存变动的统一入口一个函数管住所有加减有了流水表接下来要保证「所有库存变动都走同一个入口」。如果入库、出库、盘点各写各的 SQL迟早有人绕过校验直接改数。我一般会封装一个apply_stock_change函数参数是 SKU、仓库、变动类型、数量、单据号、操作人内部做三件事写流水、更新快照、返回最新库存。下面用 Python SQLAlchemy 写一个最小实现。from sqlalchemy import text from datetime import datetime # 变动类型到方向的映射1 表示增加库存-1 表示减少 DIRECTION {1: 1, 2: -1, 3: 1, 4: -1, 5: 1, 6: -1} def apply_stock_change(conn, sku_id, warehouse_id, txn_type, qty, ref_no, operator): 所有库存变动的唯一入口必须在事务内调用 if qty 0: raise ValueError(qty 必须为正数方向由 txn_type 决定) direction DIRECTION[txn_type] # 1. 写流水只增不改 conn.execute(text( INSERT INTO stock_txn(sku_id, warehouse_id, txn_type, qty, ref_no, operator) VALUES (:sku, :wh, :type, :qty, :ref, :op) ), {sku: sku_id, wh: warehouse_id, type: txn_type, qty: qty, ref: ref_no, op: operator}) # 2. 更新快照出库时用条件更新防止扣成负数 if direction 0: result conn.execute(text( UPDATE stock_snapshot SET qty qty - :qty WHERE sku_id :sku AND warehouse_id :wh AND qty :qty ), {qty: qty, sku: sku_id, wh: warehouse_id}) if result.rowcount 0: raise RuntimeError(库存不足出库失败) else: conn.execute(text( INSERT INTO stock_snapshot(sku_id, warehouse_id, qty) VALUES (:sku, :wh, :qty) ON DUPLICATE KEY UPDATE qty qty :qty ), {sku: sku_id, wh: warehouse_id, qty: qty}) # 3. 返回最新库存 row conn.execute(text( SELECT qty FROM stock_snapshot WHERE sku_id:sku AND warehouse_id:wh ), {sku: sku_id, wh: warehouse_id}).fetchone() return row.qty逻辑说明出库走的是UPDATE ... WHERE qty :qty这种条件更新把「判断库存够不够」和「扣减」合并成一条原子语句避免先查后改的竞态。入库用INSERT ... ON DUPLICATE KEY UPDATE第一次入库自动建快照行。参数上ref_no建议用业务单据号如订单号、入库单号排查问题时能一键串起所有相关流水。整个函数必须在数据库事务里调用否则流水写了、快照没更新数据就裂了。3. 接口与业务流程把入库、出库、盘点串成闭环3.1 三个核心接口的最小实现数据模型立住之后对外只需要暴露三个动作入库、出库、盘点。每个接口都调用上一节的apply_stock_change不自己写 SQL。下面用 FastAPI 写三个接口请求体用 Pydantic 校验能直接跑。from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from sqlalchemy import create_engine app FastAPI() engine create_engine(mysqlpymysql://user:pwdlocalhost/wms) class StockIn(BaseModel): sku_id: int warehouse_id: int qty: int Field(gt0, description入库数量必须为正) ref_no: str operator: str class StockOut(StockIn): pass app.post(/stock/in) def stock_in(body: StockIn): with engine.begin() as conn: # begin() 自动开事务异常自动回滚 qty apply_stock_change(conn, body.sku_id, body.warehouse_id, 1, body.qty, body.ref_no, body.operator) return {sku_id: body.sku_id, qty: qty} app.post(/stock/out) def stock_out(body: StockOut): try: with engine.begin() as conn: qty apply_stock_change(conn, body.sku_id, body.warehouse_id, 2, body.qty, body.ref_no, body.operator) except RuntimeError as e: raise HTTPException(status_code409, detailstr(e)) return {sku_id: body.sku_id, qty: qty}逻辑说明engine.begin()是 SQLAlchemy 的事务上下文块内抛异常会自动回滚这是保证「流水和快照同生共死」的关键。出库把库存不足转成 HTTP 409前端能明确区分「参数错」和「库存不够」。参数上Field(gt0)在入口就挡掉零和负数别指望数据库层兜底。盘点接口可以复用入库/出库用txn_type5/6区分盘盈盘亏ref_no填盘点单号。3.2 单据状态机为什么出库要先锁单再扣减真实业务里出库往往不是「点一下扣一次」而是先创建出库单草稿审核通过后才真正扣库存。如果审核和扣减之间没有状态约束同一张单被点两次「确认」库存就被扣两遍。解决办法是给出库单加状态机draft → confirmed → done只有confirmed状态才能触发扣减且扣减成功后立刻置为done。-- 出库单表用状态字段 唯一约束防重复扣减 CREATE TABLE outbound_order ( id BIGINT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(64) NOT NULL UNIQUE, status TINYINT NOT NULL DEFAULT 0 COMMENT 0草稿 1已确认 2已完成 3已取消, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;扣减时用一条带状态条件的更新来「抢单」UPDATE outbound_order SET status2 WHERE order_no? AND status1rowcount1才继续扣库存否则说明已被处理过。这个模式在并发下非常稳比加分布式锁轻量得多。参数上order_no加唯一索引从数据库层杜绝重复单据。注意状态机的状态值一旦上线就不要随意改数字含义否则历史数据全乱。要扩展就往后加比如 4 表示「部分出库」别插在中间。4. 并发与一致性超卖是怎么发生的怎么堵住4.1 先查后改的经典竞态超卖最典型的写法是先SELECT qty FROM stock_snapshot WHERE ...在代码里判断qty n再UPDATE ... SET qty qty - n。两个请求同时查到 10 件都判断够都去扣最后扣成 -10。这就是「先查后改」的竞态单机压测不一定复现一上量必翻车。堵住它有三种常见手段按推荐度排序一是条件更新上一节用的WHERE qty :qty把判断和扣减合成原子操作最简单也最可靠二是SELECT ... FOR UPDATE行锁适合扣减前还要做复杂校验的场景但锁持有时间长容易死锁三是乐观锁版本号适合冲突不激烈的场景。绝大多数库存场景条件更新就够了别过度设计。4.2 用压测验证并发安全写完不能只靠肉眼得压一下。下面用 Python 起 50 个线程同时对一个库存 100 的 SKU 出库 1 件看最终库存是不是 0、有没有负库存。import threading, requests URL http://localhost:8000/stock/out payload {sku_id: 1, warehouse_id: 1, qty: 1, ref_no: test, operator: tester} results [] def worker(): r requests.post(URL, jsonpayload) results.append(r.status_code) threads [threading.Thread(targetworker) for _ in range(150)] for t in threads: t.start() for t in threads: t.join() print(成功:, results.count(200), 库存不足:, results.count(409))逻辑说明初始库存设 100发 150 个请求正确结果是 100 个 200、50 个 409且数据库里qty恰好为 0。如果出现负数说明条件更新没生效或事务没包住。参数上线程数要大于库存数才能压出边界ref_no这里故意用同一个值是为了验证「同一单据重复提交」是否被状态机挡住——如果没挡成功数会超过 100。提示压测前把连接池调大否则瓶颈会落在连接等待上测不出真正的并发问题。MySQL 默认max_connections是 15150 线程够用但生产环境要按实际调。5. 避坑与排查上线后最常翻车的五个地方5.1 流水和快照对不上现象stock_snapshot.qty和SUM(stock_txn)算出来的值不一致。原因某次变动只写了流水没更新快照或反过来通常是事务没包全或者有人绕过统一入口直接改表。解决写一个对账脚本定期跑SELECT sku_id, warehouse_id, SUM(...) FROM stock_txn GROUP BY ...和快照比对不一致就以流水为准重算快照同时排查代码里有没有裸 SQL 改库存。5.2 出库扣成负数现象库存出现负值。原因用了先查后改或者条件更新的WHERE条件写漏了qty :qty。解决所有扣减必须走条件更新并在数据库层给stock_snapshot.qty加CHECK (qty 0)MySQL 8.0.16 支持双保险。已经出现负数的用流水重算修正。5.3 同一单据重复扣减现象一张出库单被扣了两次库存。原因接口没做幂等用户连点或重试导致重复提交。解决单据号加唯一索引扣减前用状态条件更新抢单rowcount0直接返回「已处理」。这是最容易被忽略、又最容易被用户触发的问题。5.4 盘点时把在途库存算进去现象盘点数和系统数总差一点。原因盘点时没冻结库存盘点期间还在出库入库。解决盘点单创建后对该仓库加「盘点中」标记期间拒绝出入库或记录盘点基准时间点只统计该时间点之前的流水。别在盘点时还让业务照常跑。5.5 时间字段用本地时间导致对账错乱现象跨时区或跨天对账时流水时间对不上。原因created_at用了服务器本地时间多台机器时区不一致。解决统一存 UTC展示时再转本地时区。参数上MySQL 连接串加?timezone00:00应用层统一用 UTC 生成时间。6. 进阶技巧用流水重算库存给自己留一颗后悔药系统跑久了快照和流水迟早会因为各种原因漂移这时候最值钱的能力不是「不出错」而是「能一键修回来」。我一般会写一个重算脚本以流水为唯一事实来源把任意 SKU仓库的快照重算一遍。它既是修复工具也是验证工具——每次上线新功能后跑一遍能提前发现数据不一致。def rebuild_snapshot(conn, sku_idNone, warehouse_idNone): 以 stock_txn 为准重算 stock_snapshot where, params [], {} if sku_id: where.append(sku_id :sku); params[sku] sku_id if warehouse_id: where.append(warehouse_id :wh); params[wh] warehouse_id cond (WHERE AND .join(where)) if where else # 按方向和类型汇总盘亏/出库为负 rows conn.execute(text(f SELECT sku_id, warehouse_id, SUM(CASE WHEN txn_type IN (1,3,5) THEN qty ELSE -qty END) AS qty FROM stock_txn {cond} GROUP BY sku_id, warehouse_id ), params).fetchall() for r in rows: conn.execute(text( INSERT INTO stock_snapshot(sku_id, warehouse_id, qty) VALUES (:sku, :wh, :qty) ON DUPLICATE KEY UPDATE qty :qty ), {sku: r.sku_id, wh: r.warehouse_id, qty: r.qty}) return len(rows)逻辑说明CASE WHEN txn_type IN (1,3,5)把入库、调拨入、盘盈算作增加其余算减少方向和apply_stock_change里的DIRECTION必须保持一致改一处要同步改另一处这是最容易埋雷的地方。参数上sku_id和warehouse_id都可选不传就全量重算适合夜间跑。重算前建议先备份快照表虽然流水是准的但万一流水本身被误删还有后悔药。验证方法很简单重算前后各跑一次对账脚本正常情况下重算后差异应该归零如果重算后还有差异说明流水本身有问题得往上游查。我习惯把重算脚本挂成定时任务每周日凌晨跑一次全量平时只跑增量对账。这套东西不复杂但能让你在半夜被叫起来时五分钟内把数据修好而不是对着屏幕干瞪眼。踩过最深的坑是早期图省事在商品表上直接存库存结果一次促销超卖了两百多单赔了不少钱才改成流水模型。从那以后我定了个规矩任何库存变动先问一句「流水写了吗」写不了流水的需求一律不接。希望帮到你。本文还有配套的精品资源点击获取 从Excel到WMS：简易完整仓库管理系统的数据模型与并发防超卖实战
简介这份仓库管理系统源码面向中小企业信息化负责人、物流仓储方向的开发者与学习者源自多年ERP项目实施经验将商业级WMS功能剥离精简而成聚焦小型物流仓储供应链流程支持跨平台、一处编码多处使用适合作为二次开发模板或教学范例。压缩包共445个文件约1.69MB以198个C#后端服务代码、105个Vue前端页面、86个TypeScript脚本为主体另含项目工程文件、配置与部署脚本、数据库文件及少量图片与说明文档覆盖入库、出库、库存、调拨等核心业务模块目录结构清晰便于按模块检索与复用。目前已有654人学习下载。读者可从中获得一套可直接运行的完整系统骨架理解WMS与ERP的功能边界划分参考服务层与前端页面的组织方式并借助现成配置快速搭建本地环境为自研仓储系统或课程设计提供可落地的实现思路。1. 从一张 Excel 台账说起简易完整的仓库管理系统到底该长什么样很多团队第一次做仓库管理都是从一张 Excel 台账开始的。入库加一行出库减一行月底对不上就翻聊天记录。等到 SKU 过千、多人同时改表、退货和调拨混在一起这张表就开始崩库存数对不上、谁改的查不到、超卖和积压同时出现。所谓「简易完整的仓库管理系统」不是把 ERP 那套几十个模块全搬过来而是用最小的一套数据模型和操作流程把「货在哪、有多少、怎么动的、谁动的」这四件事讲清楚并且能自己部署、自己改。它适合三类人一是中小电商或门店SKU 在几百到几万之间用不起也养不起重型系统二是做内部工具的后端或全栈工程师需要一套能跑通、能扩展的库存底座三是想理解 WMS 本质的学生和转行者需要一个能读完全部代码的完整样本。核心诉求就一句话进销存数据一致、操作可追溯、部署不折腾。下面按「数据模型怎么立 → 接口怎么写 → 并发怎么防 → 坑在哪 → 怎么验证」的顺序把一套能落地的方案讲透。2. 数据模型与表结构库存账不能只存一个数字2.1 为什么「库存数量」字段是万恶之源新手最容易犯的错是在商品表上直接放一个stock字段入库stock stock n出库stock stock - n。单机单人没问题一旦并发或需要追溯立刻翻车你只知道现在剩多少不知道这 100 件是怎么变成 80 件的退货、盘盈盘亏、调拨全糊在一起对账时就是一笔糊涂账。正确做法是引入「库存流水inventory transaction」作为唯一事实来源商品表上的库存数只是流水汇总出来的缓存。每一次库存变动都写一条流水记录变动类型、数量、方向、关联单据、操作人、时间。这样任何时刻的库存都能通过流水重算出来出问题能定位到具体哪一笔。这是 WMS 和「Excel 记账」最本质的区别也是后面所有功能的地基。常见做法是四张核心表商品表sku、仓库表warehouse、库存流水表stock_txn、库存快照表stock_snapshot可选用于加速查询。下面给出可直接建表的 SQL字段命名和类型都按 MySQL 8 的习惯来。-- 商品表只存商品静态信息不放实时库存 CREATE TABLE sku ( id BIGINT PRIMARY KEY AUTO_INCREMENT, sku_code VARCHAR(64) NOT NULL UNIQUE COMMENT 商品编码业务唯一键, name VARCHAR(128) NOT NULL, unit VARCHAR(16) DEFAULT 件, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 库存流水表唯一事实来源只增不改 CREATE TABLE stock_txn ( id BIGINT PRIMARY KEY AUTO_INCREMENT, sku_id BIGINT NOT NULL, warehouse_id BIGINT NOT NULL, txn_type TINYINT NOT NULL COMMENT 1入库 2出库 3调拨入 4调拨出 5盘盈 6盘亏, qty INT NOT NULL COMMENT 正数方向由 txn_type 决定, ref_no VARCHAR(64) COMMENT 关联单据号用于追溯, operator VARCHAR(64) NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, KEY idx_sku_wh (sku_id, warehouse_id, created_at) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 库存快照表按 SKU仓库 汇总读多写少时加速 CREATE TABLE stock_snapshot ( sku_id BIGINT NOT NULL, warehouse_id BIGINT NOT NULL, qty INT NOT NULL DEFAULT 0, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (sku_id, warehouse_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;逻辑说明stock_txn只插入不更新天然可审计txn_type用整数枚举而不是字符串索引更小、比较更快。参数上qty统一存正数方向交给txn_type判断避免正负号在业务代码里到处判断导致符号错误。idx_sku_wh这个联合索引是给「查某商品在某仓库的流水」用的顺序不能反否则范围查询用不上。提示如果 SKU 量不大几万以内可以不加stock_snapshot直接对stock_txn做SUM聚合逻辑更简单、更不容易出现快照和流水不一致。快照是性能优化不是必需品别一上来就引入。2.2 库存变动的统一入口一个函数管住所有加减有了流水表接下来要保证「所有库存变动都走同一个入口」。如果入库、出库、盘点各写各的 SQL迟早有人绕过校验直接改数。我一般会封装一个apply_stock_change函数参数是 SKU、仓库、变动类型、数量、单据号、操作人内部做三件事写流水、更新快照、返回最新库存。下面用 Python SQLAlchemy 写一个最小实现。from sqlalchemy import text from datetime import datetime # 变动类型到方向的映射1 表示增加库存-1 表示减少 DIRECTION {1: 1, 2: -1, 3: 1, 4: -1, 5: 1, 6: -1} def apply_stock_change(conn, sku_id, warehouse_id, txn_type, qty, ref_no, operator): 所有库存变动的唯一入口必须在事务内调用 if qty 0: raise ValueError(qty 必须为正数方向由 txn_type 决定) direction DIRECTION[txn_type] # 1. 写流水只增不改 conn.execute(text( INSERT INTO stock_txn(sku_id, warehouse_id, txn_type, qty, ref_no, operator) VALUES (:sku, :wh, :type, :qty, :ref, :op) ), {sku: sku_id, wh: warehouse_id, type: txn_type, qty: qty, ref: ref_no, op: operator}) # 2. 更新快照出库时用条件更新防止扣成负数 if direction 0: result conn.execute(text( UPDATE stock_snapshot SET qty qty - :qty WHERE sku_id :sku AND warehouse_id :wh AND qty :qty ), {qty: qty, sku: sku_id, wh: warehouse_id}) if result.rowcount 0: raise RuntimeError(库存不足出库失败) else: conn.execute(text( INSERT INTO stock_snapshot(sku_id, warehouse_id, qty) VALUES (:sku, :wh, :qty) ON DUPLICATE KEY UPDATE qty qty :qty ), {sku: sku_id, wh: warehouse_id, qty: qty}) # 3. 返回最新库存 row conn.execute(text( SELECT qty FROM stock_snapshot WHERE sku_id:sku AND warehouse_id:wh ), {sku: sku_id, wh: warehouse_id}).fetchone() return row.qty逻辑说明出库走的是UPDATE ... WHERE qty :qty这种条件更新把「判断库存够不够」和「扣减」合并成一条原子语句避免先查后改的竞态。入库用INSERT ... ON DUPLICATE KEY UPDATE第一次入库自动建快照行。参数上ref_no建议用业务单据号如订单号、入库单号排查问题时能一键串起所有相关流水。整个函数必须在数据库事务里调用否则流水写了、快照没更新数据就裂了。3. 接口与业务流程把入库、出库、盘点串成闭环3.1 三个核心接口的最小实现数据模型立住之后对外只需要暴露三个动作入库、出库、盘点。每个接口都调用上一节的apply_stock_change不自己写 SQL。下面用 FastAPI 写三个接口请求体用 Pydantic 校验能直接跑。from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from sqlalchemy import create_engine app FastAPI() engine create_engine(mysqlpymysql://user:pwdlocalhost/wms) class StockIn(BaseModel): sku_id: int warehouse_id: int qty: int Field(gt0, description入库数量必须为正) ref_no: str operator: str class StockOut(StockIn): pass app.post(/stock/in) def stock_in(body: StockIn): with engine.begin() as conn: # begin() 自动开事务异常自动回滚 qty apply_stock_change(conn, body.sku_id, body.warehouse_id, 1, body.qty, body.ref_no, body.operator) return {sku_id: body.sku_id, qty: qty} app.post(/stock/out) def stock_out(body: StockOut): try: with engine.begin() as conn: qty apply_stock_change(conn, body.sku_id, body.warehouse_id, 2, body.qty, body.ref_no, body.operator) except RuntimeError as e: raise HTTPException(status_code409, detailstr(e)) return {sku_id: body.sku_id, qty: qty}逻辑说明engine.begin()是 SQLAlchemy 的事务上下文块内抛异常会自动回滚这是保证「流水和快照同生共死」的关键。出库把库存不足转成 HTTP 409前端能明确区分「参数错」和「库存不够」。参数上Field(gt0)在入口就挡掉零和负数别指望数据库层兜底。盘点接口可以复用入库/出库用txn_type5/6区分盘盈盘亏ref_no填盘点单号。3.2 单据状态机为什么出库要先锁单再扣减真实业务里出库往往不是「点一下扣一次」而是先创建出库单草稿审核通过后才真正扣库存。如果审核和扣减之间没有状态约束同一张单被点两次「确认」库存就被扣两遍。解决办法是给出库单加状态机draft → confirmed → done只有confirmed状态才能触发扣减且扣减成功后立刻置为done。-- 出库单表用状态字段 唯一约束防重复扣减 CREATE TABLE outbound_order ( id BIGINT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(64) NOT NULL UNIQUE, status TINYINT NOT NULL DEFAULT 0 COMMENT 0草稿 1已确认 2已完成 3已取消, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;扣减时用一条带状态条件的更新来「抢单」UPDATE outbound_order SET status2 WHERE order_no? AND status1rowcount1才继续扣库存否则说明已被处理过。这个模式在并发下非常稳比加分布式锁轻量得多。参数上order_no加唯一索引从数据库层杜绝重复单据。注意状态机的状态值一旦上线就不要随意改数字含义否则历史数据全乱。要扩展就往后加比如 4 表示「部分出库」别插在中间。4. 并发与一致性超卖是怎么发生的怎么堵住4.1 先查后改的经典竞态超卖最典型的写法是先SELECT qty FROM stock_snapshot WHERE ...在代码里判断qty n再UPDATE ... SET qty qty - n。两个请求同时查到 10 件都判断够都去扣最后扣成 -10。这就是「先查后改」的竞态单机压测不一定复现一上量必翻车。堵住它有三种常见手段按推荐度排序一是条件更新上一节用的WHERE qty :qty把判断和扣减合成原子操作最简单也最可靠二是SELECT ... FOR UPDATE行锁适合扣减前还要做复杂校验的场景但锁持有时间长容易死锁三是乐观锁版本号适合冲突不激烈的场景。绝大多数库存场景条件更新就够了别过度设计。4.2 用压测验证并发安全写完不能只靠肉眼得压一下。下面用 Python 起 50 个线程同时对一个库存 100 的 SKU 出库 1 件看最终库存是不是 0、有没有负库存。import threading, requests URL http://localhost:8000/stock/out payload {sku_id: 1, warehouse_id: 1, qty: 1, ref_no: test, operator: tester} results [] def worker(): r requests.post(URL, jsonpayload) results.append(r.status_code) threads [threading.Thread(targetworker) for _ in range(150)] for t in threads: t.start() for t in threads: t.join() print(成功:, results.count(200), 库存不足:, results.count(409))逻辑说明初始库存设 100发 150 个请求正确结果是 100 个 200、50 个 409且数据库里qty恰好为 0。如果出现负数说明条件更新没生效或事务没包住。参数上线程数要大于库存数才能压出边界ref_no这里故意用同一个值是为了验证「同一单据重复提交」是否被状态机挡住——如果没挡成功数会超过 100。提示压测前把连接池调大否则瓶颈会落在连接等待上测不出真正的并发问题。MySQL 默认max_connections是 15150 线程够用但生产环境要按实际调。5. 避坑与排查上线后最常翻车的五个地方5.1 流水和快照对不上现象stock_snapshot.qty和SUM(stock_txn)算出来的值不一致。原因某次变动只写了流水没更新快照或反过来通常是事务没包全或者有人绕过统一入口直接改表。解决写一个对账脚本定期跑SELECT sku_id, warehouse_id, SUM(...) FROM stock_txn GROUP BY ...和快照比对不一致就以流水为准重算快照同时排查代码里有没有裸 SQL 改库存。5.2 出库扣成负数现象库存出现负值。原因用了先查后改或者条件更新的WHERE条件写漏了qty :qty。解决所有扣减必须走条件更新并在数据库层给stock_snapshot.qty加CHECK (qty 0)MySQL 8.0.16 支持双保险。已经出现负数的用流水重算修正。5.3 同一单据重复扣减现象一张出库单被扣了两次库存。原因接口没做幂等用户连点或重试导致重复提交。解决单据号加唯一索引扣减前用状态条件更新抢单rowcount0直接返回「已处理」。这是最容易被忽略、又最容易被用户触发的问题。5.4 盘点时把在途库存算进去现象盘点数和系统数总差一点。原因盘点时没冻结库存盘点期间还在出库入库。解决盘点单创建后对该仓库加「盘点中」标记期间拒绝出入库或记录盘点基准时间点只统计该时间点之前的流水。别在盘点时还让业务照常跑。5.5 时间字段用本地时间导致对账错乱现象跨时区或跨天对账时流水时间对不上。原因created_at用了服务器本地时间多台机器时区不一致。解决统一存 UTC展示时再转本地时区。参数上MySQL 连接串加?timezone00:00应用层统一用 UTC 生成时间。6. 进阶技巧用流水重算库存给自己留一颗后悔药系统跑久了快照和流水迟早会因为各种原因漂移这时候最值钱的能力不是「不出错」而是「能一键修回来」。我一般会写一个重算脚本以流水为唯一事实来源把任意 SKU仓库的快照重算一遍。它既是修复工具也是验证工具——每次上线新功能后跑一遍能提前发现数据不一致。def rebuild_snapshot(conn, sku_idNone, warehouse_idNone): 以 stock_txn 为准重算 stock_snapshot where, params [], {} if sku_id: where.append(sku_id :sku); params[sku] sku_id if warehouse_id: where.append(warehouse_id :wh); params[wh] warehouse_id cond (WHERE AND .join(where)) if where else # 按方向和类型汇总盘亏/出库为负 rows conn.execute(text(f SELECT sku_id, warehouse_id, SUM(CASE WHEN txn_type IN (1,3,5) THEN qty ELSE -qty END) AS qty FROM stock_txn {cond} GROUP BY sku_id, warehouse_id ), params).fetchall() for r in rows: conn.execute(text( INSERT INTO stock_snapshot(sku_id, warehouse_id, qty) VALUES (:sku, :wh, :qty) ON DUPLICATE KEY UPDATE qty :qty ), {sku: r.sku_id, wh: r.warehouse_id, qty: r.qty}) return len(rows)逻辑说明CASE WHEN txn_type IN (1,3,5)把入库、调拨入、盘盈算作增加其余算减少方向和apply_stock_change里的DIRECTION必须保持一致改一处要同步改另一处这是最容易埋雷的地方。参数上sku_id和warehouse_id都可选不传就全量重算适合夜间跑。重算前建议先备份快照表虽然流水是准的但万一流水本身被误删还有后悔药。验证方法很简单重算前后各跑一次对账脚本正常情况下重算后差异应该归零如果重算后还有差异说明流水本身有问题得往上游查。我习惯把重算脚本挂成定时任务每周日凌晨跑一次全量平时只跑增量对账。这套东西不复杂但能让你在半夜被叫起来时五分钟内把数据修好而不是对着屏幕干瞪眼。踩过最深的坑是早期图省事在商品表上直接存库存结果一次促销超卖了两百多单赔了不少钱才改成流水模型。从那以后我定了个规矩任何库存变动先问一句「流水写了吗」写不了流水的需求一律不接。希望帮到你。本文还有配套的精品资源点击获取