PAYATHON 2026

SQLAlchemy 和 PostgreSQL 中并发预留库存时如何锁定 FEFO 批次？

技术老齐

结论

可以使用 SELECT ... FOR UPDATE，先锁住候选库存批次，再根据锁定查询返回的余额计算分配数量。批次余额更新、预留明细和幂等记录必须放在同一个事务中。库存不足时抛出异常，回滚整个事务。

分配多个批次时，应采用固定顺序，例如 ORDER BY expires_on, id。不要默认使用 SKIP LOCKED，因为它会跳过其他事务正在锁定的早到期批次，改变 FEFO 的分配结果。

审查实现时应检查这些条件，但仅凭问题描述，还无法判断现有系统是否存在超卖。

为什么必须先锁行，再计算可用量

假设某个批次的实际库存为 10，已预留量为 0。两个事务通过普通查询都读到可用量为 10，随后各自预留 8，就可能分配出超过库存的数量。

使用行锁时，执行过程如下：

  1. 事务 A 锁住批次，计算并预留 8。
  2. 事务 B 的锁定查询等待 A 结束。
  3. A 提交后，B 根据更新后的余额计算，看到的剩余可用量为 2。

这里假设使用 PostgreSQL 的 READ COMMITTED 隔离级别。在这个级别下，锁定查询等待并发更新结束后，会重新检查该行是否符合查询条件，并返回符合条件的更新后版本。PostgreSQL 事务隔离文档

行锁应加在计算可用量所依据的余额行上。如果同一批次按仓库、库位分别保存余额，就应锁住对应的余额行。只锁批次元数据，无法保护另一张表里的数量。

示例的业务前提

可用量按以下公式计算：

available = on_hand - reserved

预留时增加 reserved，不减少 on_hand。实际出库时，再扣减实物库存并核销预留。

示例采用以下业务约定：

  • 数量使用整数基本单位；需要小数时，改用 Numeric 和 Decimal。
  • 每条订单行只允许完成一次全额预留，order_line_id 就是业务幂等键。
  • 批次到期日不可并发修改，同一天到期的批次按 id 排序。
  • expires_on >= stock_date 表示可分配，是否允许到期当天出库应按业务调整。
  • 分配范围以候选查询可见的库存为准；更严格的并发 FEFO 要求见后文。

表结构

from datetime import date

from sqlalchemy import (
    BigInteger,
    CheckConstraint,
    Date,
    ForeignKey,
    Index,
    select,
)
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    sessionmaker,
)


class Base(DeclarativeBase):
    pass


class StockBatch(Base):
    __tablename__ = "stock_batch"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    warehouse_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    sku_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    expires_on: Mapped[date] = mapped_column(Date, nullable=False)
    on_hand: Mapped[int] = mapped_column(BigInteger, nullable=False)
    reserved: Mapped[int] = mapped_column(
        BigInteger, nullable=False, server_default="0"
    )

    __table_args__ = (
        CheckConstraint("on_hand >= 0", name="ck_batch_on_hand"),
        CheckConstraint("reserved >= 0", name="ck_batch_reserved"),
        CheckConstraint(
            "reserved <= on_hand", name="ck_batch_reserved_limit"
        ),
        Index(
            "ix_batch_fefo",
            "warehouse_id", "sku_id", "expires_on", "id",
        ),
    )


class ReservationRequest(Base):
    __tablename__ = "reservation_request"

    # 一条订单行只允许完成一次预留。
    order_line_id: Mapped[int] = mapped_column(
        BigInteger, primary_key=True, autoincrement=False
    )
    warehouse_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    sku_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    qty: Mapped[int] = mapped_column(BigInteger, nullable=False)
    stock_date: Mapped[date] = mapped_column(Date, nullable=False)

    __table_args__ = (
        CheckConstraint("qty > 0", name="ck_request_qty"),
    )


class Reservation(Base):
    __tablename__ = "reservation"

    order_line_id: Mapped[int] = mapped_column(
        ForeignKey("reservation_request.order_line_id"),
        primary_key=True,
    )
    batch_id: Mapped[int] = mapped_column(
        ForeignKey("stock_batch.id"),
        primary_key=True,
    )
    qty: Mapped[int] = mapped_column(BigInteger, nullable=False)

    __table_args__ = (
        CheckConstraint("qty > 0", name="ck_reservation_qty"),
    )

Reservation 的联合主键保证同一订单行在同一批次下只有一条明细。实际 ERP 中，还应为 order_line_id、warehouse_id 和 sku_id 添加指向对应业务表的外键；示例省略了这些业务表。

在事务内完成幂等检查与多批次分配

class InsufficientStock(Exception):
    pass


class IdempotencyConflict(Exception):
    pass


def read_allocations(session: Session, order_line_id: int) -> list[dict]:
    rows = session.execute(
        select(Reservation.batch_id, Reservation.qty)
        .where(Reservation.order_line_id == order_line_id)
        .order_by(Reservation.batch_id)
    )
    return [dict(row._mapping) for row in rows]


def reserve_in_transaction(
    session: Session,
    *,
    order_line_id: int,
    warehouse_id: int,
    sku_id: int,
    qty: int,
    stock_date: date,
) -> list[dict]:
    if isinstance(qty, bool) or not isinstance(qty, int) or qty <= 0:
        raise ValueError("qty must be a positive integer")

    # 先通过唯一键取得这次业务操作的执行权。
    created = session.execute(
        insert(ReservationRequest)
        .values(
            order_line_id=order_line_id,
            warehouse_id=warehouse_id,
            sku_id=sku_id,
            qty=qty,
            stock_date=stock_date,
        )
        .on_conflict_do_nothing(
            index_elements=[ReservationRequest.order_line_id]
        )
        .returning(ReservationRequest.order_line_id)
    ).scalar_one_or_none()

    if created is None:
        # READ COMMITTED 下，这是一个新的查询快照。
        previous = session.execute(
            select(ReservationRequest).where(
                ReservationRequest.order_line_id == order_line_id
            )
        ).scalar_one()

        if (
            previous.warehouse_id,
            previous.sku_id,
            previous.qty,
            previous.stock_date,
        ) != (warehouse_id, sku_id, qty, stock_date):
            raise IdempotencyConflict(
                "order line already reserved with different parameters"
            )

        return read_allocations(session, order_line_id)

    # 一次取回并锁住所有符合条件的候选批次。
    batches = session.scalars(
        select(StockBatch)
        .where(
            StockBatch.warehouse_id == warehouse_id,
            StockBatch.sku_id == sku_id,
            StockBatch.expires_on >= stock_date,
            StockBatch.on_hand > StockBatch.reserved,
        )
        .order_by(StockBatch.expires_on, StockBatch.id)
        .with_for_update(of=StockBatch)
        .execution_options(populate_existing=True)
    ).all()

    remaining = qty
    plan = []

    # 这里只计算分配方案，尚未修改余额。
    for batch in batches:
        take = min(batch.on_hand - batch.reserved, remaining)
        if take > 0:
            plan.append((batch, take))
            remaining -= take
        if remaining == 0:
            break

    if remaining:
        raise InsufficientStock(f"short by {remaining}")

    for batch, take in plan:
        batch.reserved += take
        session.add(
            Reservation(
                order_line_id=order_line_id,
                batch_id=batch.id,
                qty=take,
            )
        )

    session.flush()
    return read_allocations(session, order_line_id)

with_for_update() 生成锁定查询。populate_existing=True 会刷新 Session 中可能已经加载过的对象，避免继续使用旧的 ORM 属性值。SQLAlchemy 查询执行选项

同一订单行的两个并发请求通过幂等记录的唯一键协调执行。前一个事务提交后，后一个请求进入已有结果分支；前一个事务回滚后，后一个请求可以成功插入并继续分配。

插入后的独立查询必须保留：ON CONFLICT DO NOTHING 可能因为另一个事务的结果而放弃插入，即使该记录不在插入语句最初的快照中。PostgreSQL 对 ON CONFLICT 的说明

由外层统一提交或回滚

reserve_in_transaction() 不负责提交，调用方负责控制事务边界：

# engine 是应用已经配置好的 SQLAlchemy Engine。
reservation_engine = engine.execution_options(
    isolation_level="READ COMMITTED"
)
SessionFactory = sessionmaker(bind=reservation_engine)


def reserve_order_line(**params):
    with SessionFactory.begin() as session:
        result = reserve_in_transaction(session, **params)

        # 需要原子提交的订单状态更新，也应在这里完成。

    # 离开上下文并成功提交后，才向调用方返回成功。
    return result

sessionmaker.begin() 在正常退出时提交，在异常退出时回滚并关闭 Session。flush() 只是把变更发送给数据库，不会提交，也不会释放这里持有的行锁。SQLAlchemy 事务管理

库存不足、插入明细失败或约束检查失败时，本次事务中的幂等记录、余额更新和预留明细都会一起回滚。应在事务上下文之外将异常转换成 Flask 响应，不要在上下文内部吞掉异常后正常退出。

如果一次确认包含多个订单行，应在同一个 SessionFactory.begin() 中多次调用 reserve_in_transaction()，全部完成后再提交。逐行调用 reserve_order_line() 会形成多个独立事务，无法保证整张订单全部成功或全部失败。

传入的订单行、SKU、仓库和数量应来自服务端验证过的订单数据。如果需要防止订单在确认期间被修改，还应在同一事务中锁住订单或订单行。

多批次锁定与严格 FEFO 的边界

示例通过 .all() 先取得全部候选行的锁，再计算分配。这种实现容易审查，但也会锁住最终没有用到的候选批次。批次数量较多时，可以改为分段锁定，已经取得的锁仍须保留到事务结束。

所有库存写入路径应采用一致的锁顺序。涉及多个 SKU 或仓库时，还要统一各库存范围的处理顺序，例如先按 (warehouse_id, sku_id) 排序，再在每个范围内按 (expires_on, id) 处理。这样可以降低死锁概率，但仍须处理数据库报告的死锁。PostgreSQL 显式锁定文档

使用 SKIP LOCKED 会有两个后果：

  • 早到期批次被锁住时，可能转而使用晚到期批次。
  • 被锁住的库存不参与计算，因此可能返回库存不足，即使对方回滚后这些库存仍可使用。

只有业务明确接受这些行为时，才应使用 with_for_update(skip_locked=True)。PostgreSQL 锁定子句

行锁只保护锁到的行，整个候选集合仍可能发生变化：

  • 并发插入的新批次可能不在查询结果中。
  • 查询开始时没有可用量的批次，可能随后因释放预留而恢复可用。
  • 到期日在等待期间被修改时，返回结果可能不再按更新后的到期日排序。PostgreSQL 文档说明了 ORDER BY 配合锁定查询时的这一行为。PostgreSQL 排序与行锁注意事项

如果业务要求这些变化也与 FEFO 分配串行执行，可以为每个 (warehouse_id, sku_id) 建立一条稳定的库存范围记录。所有预留、释放、入库及到期日修改操作，都先锁住这条记录，再查询批次。同一库存范围内的操作因此会依次执行，代价是更高的锁竞争。所有相关写入入口都必须遵守这一锁定协议。

数据库约束、幂等与重试还应检查什么

示例中的约束各有作用：

  • NOT NULL 防止数量为空而绕过有效的数值检查。
  • CHECK (reserved >= 0 AND reserved <= on_hand) 防止余额超出合法范围。
  • 幂等记录的主键防止同一订单行再次执行预留。
  • 预留明细的联合主键防止同一订单行重复写入同一批次。
  • 外键防止明细引用不存在的请求或批次。

这些约束无法自动保证 reserved 等于所有有效预留明细的数量之和。PostgreSQL 的普通 CHECK 无法可靠维护这种跨行、跨表的汇总关系。释放、取消、出库等操作也必须在事务中同步维护余额与明细。如果需要由数据库进一步强制保证，可以考虑设计合适的触发器，或统一数据库写入入口。PostgreSQL 约束文档

幂等设计还需要明确操作的生命周期。示例只覆盖一次成功预留。如果允许取消后重新预留，应增加预留轮次或业务操作 ID，并保留旧操作结果，避免旧请求重试时触发新的分配。多租户系统还应将租户范围纳入业务唯一键。相同幂等键携带不同参数时，应报冲突。

约束和索引需要通过 Alembic 实际部署到数据库。仅修改 ORM 模型不会改变已有表；迁移前还应检查历史数据是否违反新增约束。

遇到死锁 40P01 或序列化失败 40001 时，应先回滚，再用新事务重试完整业务操作，并设置有限的重试次数和退避。不要只重试最后一条 SQL，也不要把所有 IntegrityError 都视为可重试错误。如果连接中断导致提交结果未知，应沿用同一业务幂等键查询或重试。

验证并发行为时，至少使用两个独立数据库连接，检查竞争同一库存、跨批次分配、数量不足回滚，以及同一幂等键并发重试的情况。核对结果时，应同时检查余额、预留明细总量和幂等记录，据此判断现有实现是否满足要求。

备注：内容仅供参考。