电子商务源码中的购物车和订单流程是怎样实现的？
购物车看起来只是“把商品放进去”，订单看起来只是“生成一张单子”，但真正的难点在于商品价格会变化、库存会被多人同时购买、用户可能重复点击付款，还可能在下单途中断网。设计不当会出现超卖、重复扣款、订单金额被篡改等问题。
本文从常见业务流程入手，说明购物车与订单的数据边界，并提供一份使用 Python 标准库和 SQLite 的可运行示例。示例适合本地学习、原型验证和理解事务；生产系统还需结合实际数据库、支付渠道、权限体系和业务规则进行扩展。
一、购物车和订单分别负责什么
购物车是可变的购买意向
购物车记录用户当前打算买什么，通常支持增减数量、删除商品、勾选结算等操作。它不是交易凭证：商品价格、库存、促销资格都可能在购物车保存期间发生变化。因此，购物车中显示的价格只能作为展示信息，不能直接作为最终扣款依据。
常见购物车形态包括：
- 登录用户购物车：保存在服务端数据库，换设备登录后仍可见。
- 访客购物车：保存在浏览器或由服务端按匿名标识保存。合并到登录购物车时，需要定义重复商品如何合并及数量上限。
- 临时购物车：用于快速下单、报价或短期会话，过期后可以清理。
浏览器本地存储适合改善界面体验，但不应作为价格、库存、用户身份或订单状态的权威来源。客户端传来的商品 ID 和数量可以作为用户意图，价格等关键字段必须由服务端重新查询和计算。
订单是交易过程的持久记录
订单记录某次购买时的商品、数量、成交价、收货信息、金额和状态。与购物车不同，订单中的商品行通常需要保存下单时快照，例如商品名称、SKU、成交单价。即使之后商品改名或调价，历史订单仍应保持原样。
常见订单相关数据包括订单主表、订单商品行、支付记录、退款记录和物流记录。实际项目是否拆表，取决于业务规模和查询需求；但状态变化、金额来源和关联关系必须清楚。
二、典型流程与关键边界
- 加入购物车：校验用户、商品是否可购买及数量范围。服务端保存商品标识和数量，必要时展示当前价格。
- 查看购物车：读取商品当前信息，标记已下架、缺货或价格变化的项目。不要把显示结果当作结算承诺。
- 提交结算：客户端提交要购买的商品及数量，并提供收货地址、优惠券等业务输入。服务端重新校验这些输入。
- 创建订单：重新读取商品价格和库存，计算优惠、运费及应付金额，在一致性边界内写入订单和订单行，并按业务策略处理库存。
- 支付：调用支付渠道创建支付请求。支付成功后，以经过验证的渠道通知或主动查询结果更新订单，不以浏览器跳转页面作为支付成功凭据。
- 履约与售后：根据订单状态发货、取消、退款。状态迁移应经过校验，不能允许任意状态被客户端直接设置。
这里有一个重要边界：购物车操作通常允许失败后重试；下单和支付则必须考虑重复请求。用户点两次按钮、客户端超时后重发请求，都不应意外生成两笔订单或造成重复扣款。
三、数据设计：金额、快照与状态
金额用整数最小货币单位保存
不要用二进制浮点数直接表示货币。许多十进制小数无法被浮点数精确表示，反复计算可能产生精度误差。常见做法是用整数表示最小货币单位，例如人民币以“分”为单位：12.34 元保存为 1234。若业务包含多种币种，应同时保存币种，并遵循该币种的精度规则。
订单行保存购买时快照
订单商品行至少应能还原成交内容。下例保存商品 ID、名称快照、单价（分）和数量。复杂业务还可能保存 SKU 属性、税费、优惠分摊、供应商信息等。快照并不意味着复制所有商品数据，而是保留解释订单金额和履约所需的关键字段。
状态由服务端驱动
订单状态可以设计为待支付、已支付、待发货、已发货、已完成、已取消、退款中等。具体状态取决于业务，不存在适用于所有商城的固定枚举。应明确定义每种状态的合法迁移、触发方、超时规则以及库存和退款影响。客户端只能发起动作，不能直接提交“把订单改成已支付”。
四、可运行示例：SQLite 中加入购物车与创建订单
下面示例只使用 Python 标准库，无需安装第三方包。将代码保存为 shop_demo.py，使用 python shop_demo.py 运行。它会创建本地数据库、加入购物车并创建订单。
示例刻意把范围控制在基础交易环节：购物车保存用户、商品和数量；下单时从商品表重新读取价格，使用事务检查并扣减库存，保存订单与订单行快照；以用户和幂等键避免同一结算请求重复创建订单。它没有实现登录、地址、优惠券、支付、退款或配送，因此不能直接作为完整商城上线。
import sqlite3
from pathlib import Path
DB_PATH = Path("shop_demo.sqlite3")
MAX_CART_QUANTITY = 99
def connect():
# isolation_level=None 让事务由代码显式控制。
conn = sqlite3.connect(DB_PATH, isolation_level=None)
conn.row_factory = sqlite3.Row
return conn
def init_db():
with connect() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
price_cents INTEGER NOT NULL CHECK (price_cents >= 0),
stock INTEGER NOT NULL CHECK (stock >= 0),
active INTEGER NOT NULL DEFAULT 1
);
CREATE TABLE IF NOT EXISTS cart_items (
user_id INTEGER NOT NULL,
product_id INTEGER NOT NULL REFERENCES products(id),
quantity INTEGER NOT NULL CHECK (quantity > 0),
PRIMARY KEY (user_id, product_id)
);
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
idempotency_key TEXT NOT NULL,
total_cents INTEGER NOT NULL CHECK (total_cents >= 0),
status TEXT NOT NULL,
UNIQUE (user_id, idempotency_key)
);
CREATE TABLE IF NOT EXISTS order_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id INTEGER NOT NULL REFERENCES orders(id),
product_id INTEGER NOT NULL,
product_name TEXT NOT NULL,
unit_price_cents INTEGER NOT NULL,
quantity INTEGER NOT NULL,
line_total_cents INTEGER NOT NULL
);
""")
# 演示数据：仅在商品表为空时插入。
count = conn.execute("SELECT COUNT(*) FROM products").fetchone()[0]
if count == 0:
conn.executemany(
"""INSERT INTO products (id, name, price_cents, stock)
VALUES (?, ?, ?, ?)""",
[
(1, "机械键盘", 25900, 10),
(2, "鼠标垫", 3900, 30),
],
)
def add_to_cart(user_id, product_id, quantity):
# bool 是 int 的子类，因此这里显式排除 bool。
if isinstance(quantity, bool) or not isinstance(quantity, int):
raise ValueError("quantity 必须是整数")
if quantity < 1 or quantity > MAX_CART_QUANTITY:
raise ValueError(f"quantity 必须在 1 到 {MAX_CART_QUANTITY} 之间")
with connect() as conn:
conn.execute("BEGIN IMMEDIATE")
try:
product = conn.execute(
"SELECT id, stock, active FROM products WHERE id = ?",
(product_id,),
).fetchone()
if product is None or not product["active"]:
raise ValueError("商品不存在或已下架")
row = conn.execute(
"""SELECT quantity FROM cart_items
WHERE user_id = ? AND product_id = ?""",
(user_id, product_id),
).fetchone()
new_quantity = quantity + (row["quantity"] if row else 0)
if new_quantity > MAX_CART_QUANTITY:
raise ValueError("购物车商品数量超过上限")
if new_quantity > product["stock"]:
raise ValueError("当前库存不足")
conn.execute(
"""INSERT INTO cart_items (user_id, product_id, quantity)
VALUES (?, ?, ?)
ON CONFLICT(user_id, product_id)
DO UPDATE SET quantity = excluded.quantity""",
(user_id, product_id, new_quantity),
)
conn.commit()
return new_quantity
except Exception:
conn.rollback()
raise
def create_order(user_id, idempotency_key):
if not isinstance(idempotency_key, str) or not idempotency_key.strip():
raise ValueError("必须提供非空 idempotency_key")
if len(idempotency_key) > 128:
raise ValueError("idempotency_key 长度不能超过 128 个字符")
with connect() as conn:
# SQLite 的写事务会串行化写入；这里用于示范简单的库存一致性。
conn.execute("BEGIN IMMEDIATE")
try:
existing = conn.execute(
"""SELECT id, total_cents, status FROM orders
WHERE user_id = ? AND idempotency_key = ?""",
(user_id, idempotency_key),
).fetchone()
if existing:
conn.commit()
return dict(existing)
cart = conn.execute(
"""SELECT c.product_id, c.quantity, p.name, p.price_cents,
p.stock, p.active
FROM cart_items AS c
JOIN products AS p ON p.id = c.product_id
WHERE c.user_id = ?
ORDER BY c.product_id""",
(user_id,),
).fetchall()
if not cart:
raise ValueError("购物车为空")
total_cents = 0
for item in cart:
if not item["active"]:
raise ValueError(f"商品已下架：{item['name']}")
if item["quantity"] > item["stock"]:
raise ValueError(f"库存不足：{item['name']}")
total_cents += item["price_cents"] * item["quantity"]
cursor = conn.execute(
"""INSERT INTO orders
(user_id, idempotency_key, total_cents, status)
VALUES (?, ?, ?, ?)""",
(user_id, idempotency_key, total_cents, "pending_payment"),
)
order_id = cursor.lastrowid
for item in cart:
# 条件更新再次检查库存；失败时整个事务回滚。
updated = conn.execute(
"""UPDATE products SET stock = stock - ?
WHERE id = ? AND active = 1 AND stock >= ?""",
(item["quantity"], item["product_id"], item["quantity"]),
)
if updated.rowcount != 1:
raise ValueError(f"库存不足：{item['name']}")
conn.execute(
"""INSERT INTO order_items
(order_id, product_id, product_name, unit_price_cents,
quantity, line_total_cents)
VALUES (?, ?, ?, ?, ?, ?)""",
(
order_id,
item["product_id"],
item["name"],
item["price_cents"],
item["quantity"],
item["price_cents"] * item["quantity"],
),
)
conn.execute("DELETE FROM cart_items WHERE user_id = ?", (user_id,))
conn.commit()
return {
"id": order_id,
"total_cents": total_cents,
"status": "pending_payment",
}
except Exception:
conn.rollback()
raise
def main():
init_db()
user_id = 1001
print("加入购物车，数量：", add_to_cart(user_id, 1, 1))
print("加入购物车，数量：", add_to_cart(user_id, 2, 2))
# 同一个用户、同一个业务请求重试时，传回相同的幂等键。
order = create_order(user_id, "checkout-demo-0001")
print("订单结果：", order)
# 再次调用会返回已有订单，不再重复扣库存或创建订单。
retry = create_order(user_id, "checkout-demo-0001")
print("重试结果：", retry)
if __name__ == "__main__":
main()五、示例中的事务与幂等分别解决什么问题
BEGIN IMMEDIATE 让 SQLite 在事务开始时取得写锁，避免两个写事务在读取库存后同时按旧库存继续扣减。下单过程中的订单创建、库存更新、订单行写入和清空购物车处于同一事务里；其中任何一步失败，都会回滚此前的数据库修改，避免留下半张订单。
SQLite 的写锁粒度和并发能力有限，这种写法适合演示，不代表生产环境的通用并发方案。生产系统常使用 PostgreSQL、MySQL 等数据库，结合事务、行级锁、条件更新、唯一约束和合适的隔离级别处理竞争。不能仅凭“先查库存，再扣库存”保证并发安全；库存扣减应当是原子操作，或在事务中获得相应锁并检查更新结果。
幂等键表示同一个业务操作的稳定标识。示例通过 UNIQUE (user_id, idempotency_key) 约束保证同一用户重复提交相同键时不会新增订单。真实接口中，键通常由客户端为一次结算尝试生成，重试时保持不变；新的购买意图则使用新键。服务端应验证键的格式和长度，并按用户或租户隔离，不能把幂等键当作登录凭证或授权令牌。
生产实现还要考虑同一幂等键被不同请求参数复用的情况。可保存请求摘要或关键输入：若键相同但请求内容不同，应拒绝请求或返回明确冲突，而不是把旧订单结果误当成新请求结果。还需规划记录保留期限，避免清理策略与客户端重试时间窗口冲突。
六、库存何时扣减：要按业务选择
示例在创建待支付订单时就扣减库存。这样实现简单，订单超时未支付时必须有任务释放库存；取消订单和支付通知并发到达时，也要保证库存只释放或确认一次。库存的保留期限、自动取消机制和状态迁移需要配套设计。
另一种做法是在下单时设置库存预占，支付成功后转为实际扣减，超时则释放预占。适合需要明确“保留一段时间”的业务，但需要额外记录预占数量、到期时间和释放状态，并处理延迟任务、重复任务及支付回调竞态。也有业务选择支付后再扣库存，但这会提高付款成功却无货的风险，必须有明确的补偿和客服策略。
秒杀、限量活动和多仓库存尤其不能照搬普通购物车逻辑。它们可能需要库存服务、原子计数、队列削峰、限购规则和防刷机制。任何实现都应明确库存的权威数据源，避免购物车、订单服务和商品服务各自维护一份互不一致的“真实库存”。
七、支付不是创建订单的一步简单调用
通常先在本地创建待支付订单，再请求支付渠道创建支付单。支付渠道返回的支付参数交给前端使用；支付成功后，渠道向服务端发送通知。服务端要验证通知签名、商户与订单信息、金额和币种，并按渠道要求处理通知重试。验证成功后，在数据库事务内把支付记录和订单状态更新到位。
通知可能重复、延迟或乱序，因此支付处理也要幂等。不能因为收到一次“支付成功”消息就无条件重复执行发货、记账或赠送权益。可通过支付渠道交易号唯一约束、订单状态校验和业务事件去重实现。对不确定的支付结果，应调用渠道查询接口核实，而不是根据前端页面参数推断。
订单服务与外部支付系统无法共享一个普通数据库事务。若本地订单写入成功但支付请求失败，或支付成功后服务端暂时未收到通知，应通过重试、主动查询、补偿任务和对账处理最终一致性。不要把网络调用放进数据库事务并长时间持锁。
八、参数限制与安全用途说明
- 数量限制：示例限制每个商品数量为 1 到 99，是演示用的业务上限，并非 Python、SQLite 或电商行业的统一限制。实际值应结合销售规则、数据库类型、库存模型和接口约定确定。
- 幂等键长度：示例限制为最多 128 个字符，是该示例的输入策略，不是通用标准。线上 API 应定义字符集、最大长度和生成方式，并与数据库字段及客户端实现保持一致。
- 类型检查：代码检查数量为整数并排除布尔值，避免 Python 中
True也被视为整数。API 层还应校验 JSON 类型、缺失字段、整数范围和非法编码。 - SQL 参数化：示例使用占位符传入值，避免把用户输入拼接进 SQL。表名、列名等 SQL 结构不能通过普通值占位符安全传入，动态结构应使用固定白名单。
- 用户身份：示例的
user_id是函数参数，仅为本地演示。线上必须从已验证的登录会话或访问令牌中取得用户身份，不能相信客户端随意提交的用户 ID。 - 支付数据：不得把客户端提交的“已支付”状态、金额或支付截图作为可靠凭证。需要按支付渠道文档验证签名、金额、币种、商户号和交易状态，并安全保存密钥。
- 隐私与权限：订单查询必须校验当前用户是否有权访问目标订单。地址、电话等个人信息应最小化收集、限制访问并按适用法规和组织规范保护。
九、常见错误及修正方向
把购物车价格直接当订单金额
客户端可修改请求参数，购物车中的价格也可能过期。下单时按商品和 SKU 从可信数据源重新取价，并在订单中保存成交快照。优惠券、会员价和运费也必须由服务端按规则计算。
金额使用浮点数
浮点数计算可能产生精度误差。金额用整数最小货币单位或合适的定点十进制类型表示，并明确折扣、税费和分摊的舍入规则。多个订单行的优惠分摊需要保证合计与订单总额一致。
只查询库存，不原子扣减
并发请求可能都读到有库存，然后都下单成功。使用事务和原子条件更新，检查受影响行数；对数据库锁冲突、死锁等情况设计有限重试，并确保重试操作本身幂等。
下单后清空整个购物车
如果用户可以多标签操作，结算过程中新增的商品可能被误删。示例简化为下单后清空该用户全部购物车，因此有意没有声称支持并发编辑。正式实现可记录本次结算的商品集合，只删除已纳入订单且数量未变化的购物车行，或使用购物车版本号检测变更。
订单行只保存商品 ID
商品名称、价格和规格未来会变化，只保存外键可能无法准确还原购买时内容。订单行应保存必要快照，同时保留商品 ID 供关联查询；商品删除策略也要避免破坏历史订单。
重复请求生成多张订单
禁用按钮只能改善界面体验，不能解决网络重试和并发请求。应在服务端使用唯一约束和幂等键，并为支付通知、取消、退款等可能重试的操作分别定义去重策略。
只在前端隐藏不可购买商品
前端显示不是权限边界。下单接口必须重新检查商品状态、库存、购买数量限制、用户资格、优惠有效期和价格规则。验证失败时应返回可解释的错误，并让用户刷新结算信息。
十、从示例走向实际项目
把示例改造成正式服务前，至少需要补齐身份认证与订单授权、收货地址快照、SKU 和多规格商品、运费与税费、优惠规则、支付渠道接入、超时取消、退款、库存恢复、日志审计和定期对账。API 应明确请求和响应结构、错误码、并发冲突处理及重试约定。
测试时不要只覆盖“正常下单”。还应覆盖空购物车、商品下架、库存不足、数量越界、价格变化、重复幂等键、相同键不同参数、两个请求同时购买最后一件商品、支付通知重复到达、取消与支付并发，以及事务中途失败后数据是否完整回滚。
购物车可以理解为用户可修改的购买意向，订单则是服务端确认后的交易记录。两者之间的结算过程，核心是重新验证、可靠计算、原子写入和可安全重试。把价格与库存的权威来源放在服务端，给订单保存必要快照，为并发和外部支付设计幂等与补偿机制，才是购物车和订单流程能够稳定运行的基础。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
