Shopify 用 MySQL + SKIP LOCKED 替代 Redis 做库存预留——一行 SQL 如何撑住百万级并发

CREATE TABLE inventory_reservations (
    sku_id        BIGINT NOT NULL,
    reservation_id BIGINT NOT NULL,
    quantity      INT NOT NULL,
    status        ENUM('pending', 'confirmed', 'released') NOT NULL DEFAULT 'pending',
    created_at    TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at    TIMESTAMP NOT NULL,
    PRIMARY KEY (sku_id, reservation_id),  -- 复合主键，sku_id 在前
    INDEX idx_expires (status, expires_at)  -- 用于过期清理
) ENGINE=InnoDB;

-- 事务开始
START TRANSACTION;

-- 查询 SKU 的可用库存，跳过正在被其他事务处理的行
SELECT available_qty
FROM inventory_stock
WHERE sku_id = 1001
FOR UPDATE SKIP LOCKED;

-- 如果返回空，说明该 SKU 正在被其他事务锁定处理，直接放弃或走降级逻辑
-- 如果返回了可用数量且足够，插入预留记录
INSERT INTO inventory_reservations
    (sku_id, reservation_id, quantity, status, expires_at)
VALUES
    (1001, 20250712001, 2, 'pending', NOW() + INTERVAL 10 MINUTE);

-- 扣减可用库存
UPDATE inventory_stock
SET available_qty = available_qty - 2
WHERE sku_id = 1001 AND available_qty >= 2;

-- 如果 UPDATE 的 affected_rows = 0，说明并发下库存已被其他人扣完
-- 回滚事务，返回"库存不足"

COMMIT;

UPDATE inventory_reservations
SET status = 'confirmed'
WHERE sku_id = 1001 AND reservation_id = 20250712001 AND status = 'pending';

-- 在独立事务中执行
START TRANSACTION;

-- 找出所有已过期的 pending 预留，跳过正在被处理中的行
SELECT sku_id, reservation_id, quantity
FROM inventory_reservations
WHERE status = 'pending' AND expires_at < NOW()
FOR UPDATE SKIP LOCKED;

-- 对每条过期记录：
-- 1. 释放库存
UPDATE inventory_stock
SET available_qty = available_qty + quantity
WHERE sku_id = ?;

-- 2. 标记预留为已释放
UPDATE inventory_reservations
SET status = 'released'
WHERE sku_id = ? AND reservation_id = ? AND status = 'pending';

COMMIT;

import pymysql
from datetime import datetime, timedelta

def reserve_inventory(sku_id: int, quantity: int, reservation_id: int):
    conn = pymysql.connect(
        host='127.0.0.1', port=3306,
        user='shop', password='shoppass', db='inventory'
    )
    try:
        with conn.cursor() as cur:
            conn.begin()

            # 查询可用库存，跳过被锁住的行
            cur.execute(
                "SELECT available_qty FROM inventory_stock "
                "WHERE sku_id = %s FOR UPDATE SKIP LOCKED",
                (sku_id,)
            )
            row = cur.fetchone()
            if row is None:
                # SKU 正在被其他事务处理，走降级逻辑
                conn.rollback()
                return {"status": "conflict", "message": "SKU 正在被处理，请重试"}

            available = row[0]
            if available < quantity:
                conn.rollback()
                return {"status": "insufficient", "message": f"可用 {available}，需要 {quantity}"}

            # 插入预留记录
            expires_at = datetime.utcnow() + timedelta(minutes=10)
            cur.execute(
                "INSERT INTO inventory_reservations "
                "(sku_id, reservation_id, quantity, status, expires_at) "
                "VALUES (%s, %s, %s, 'pending', %s)",
                (sku_id, reservation_id, quantity, expires_at)
            )

            # 扣减库存
            cur.execute(
                "UPDATE inventory_stock SET available_qty = available_qty - %s "
                "WHERE sku_id = %s AND available_qty >= %s",
                (quantity, sku_id, quantity)
            )
            if cur.rowcount == 0:
                # 并发下被其他人抢先扣完
                conn.rollback()
                return {"status": "insufficient", "message": "并发冲突，库存已被扣完"}

            conn.commit()
            return {"status": "reserved", "reservation_id": reservation_id}

    except Exception as e:
        conn.rollback()
        return {"status": "error", "message": str(e)}
    finally:
        conn.close()

CREATE TABLE inventory_stock (
    sku_id        BIGINT PRIMARY KEY,
    available_qty INT NOT NULL,
    total_qty     INT NOT NULL
) ENGINE=InnoDB;

-- 初始化测试数据
INSERT INTO inventory_stock VALUES (1001, 50, 100);