如何在PHP（PDO）中使用MySQL查询队列解决并发下单库存超卖问题

嘿，这个并发超卖的坑我之前踩过好几次，刚好你想通过队列来实现用户依次下单，那我给你拆解清楚实现思路，从适合中小团队的MySQL队列方案，到高并发场景的Redis方案都给你捋明白：

一、先搞懂为啥会出现库存变负的问题 #

本质上这是竞态条件搞的鬼：当两个用户同时下单时，他们的请求会同时执行「读取库存→判断库存足够→扣减库存」这三步。因为这三步不是原子操作，两个请求都能读到初始库存1，都觉得可以下单，最后都执行扣减，结果库存就变成-1了。

用队列的核心逻辑就是把这些并发的请求变成串行处理——同一时间只处理一个下单请求，从根源上避免两个请求同时操作库存的情况。

二、用MySQL做请求队列（不用额外加中间件） #

如果你的用户量还没到特别大的程度，直接用现有的MySQL建个队列表就行，不用额外装Redis或者MQ，成本很低：

1. 先建一个队列表 #

专门用来存待处理的下单请求：

CREATE TABLE `order_queue` (
  `id` INT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
  `user_id` INT UNSIGNED NOT NULL, -- 下单用户ID
  `product_id` INT UNSIGNED NOT NULL, -- 商品ID
  `quantity` INT NOT NULL DEFAULT 1, -- 下单数量
  `status` ENUM('pending', 'processing', 'success', 'failed') NOT NULL DEFAULT 'pending', -- 请求状态
  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, -- 提交时间
  `processed_at` DATETIME NULL -- 处理完成时间
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

2. 用户下单时，先把请求扔进队列 #

用户点击下单按钮后，不要直接去操作库存和订单，而是先把请求插入队列，给用户一个友好提示就行：

// 先初始化PDO连接（这里假设你已经有通用的数据库连接逻辑）
$pdo = new PDO('mysql:host=localhost;dbname=your_db;charset=utf8mb4', 'db_user', 'db_pass');
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

try {
    $userId = $_SESSION['user_id']; // 假设用户已登录，能拿到用户ID
    $productId = $_POST['product_id'];
    $quantity = 1; // 可以改成用户选择的数量

    // 插入队列请求
    $stmt = $pdo->prepare("INSERT INTO order_queue (user_id, product_id, quantity) VALUES (?, ?, ?)");
    $stmt->execute([$userId, $productId, $quantity]);

    // 给用户反馈
    echo "你的下单请求已提交，正在排队处理，请稍后查看订单状态~";
} catch(PDOException $e) {
    echo "提交请求失败：" . $e->getMessage();
}

3. 写个后台脚本，定时处理队列里的请求 #

你需要一个脚本，每隔一小段时间就从队列里取出最早的未处理请求，逐个处理。可以用Linux的nohup让脚本常驻后台：

<?php
// queue_processor.php - 队列处理脚本
$pdo = new PDO('mysql:host=localhost;dbname=your_db;charset=utf8mb4', 'db_user', 'db_pass');
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

while(true) {
    try {
        // 开启事务，用FOR UPDATE锁定第一条待处理请求，避免多个进程同时抢着处理
        $pdo->beginTransaction();
        $stmt = $pdo->prepare("SELECT * FROM order_queue WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1 FOR UPDATE");
        $stmt->execute();
        $queueItem = $stmt->fetch(PDO::FETCH_ASSOC);

        if (!$queueItem) {
            // 没有待处理请求，提交事务后休眠1秒再查
            $pdo->commit();
            sleep(1);
            continue;
        }

        // 先把请求标记为处理中，避免重复处理
        $stmt = $pdo->prepare("UPDATE order_queue SET status = 'processing' WHERE id = ?");
        $stmt->execute([$queueItem['id']]);
        $pdo->commit();

        // 开始处理下单逻辑
        $productId = $queueItem['product_id'];
        $quantity = $queueItem['quantity'];
        $userId = $queueItem['user_id'];

        // 再次开启事务，处理库存和订单
        $pdo->beginTransaction();

        // 查询库存时加行锁，确保拿到的是最新的库存数据
        $stmt = $pdo->prepare("SELECT stock FROM products WHERE id = ? FOR UPDATE");
        $stmt->execute([$productId]);
        $product = $stmt->fetch(PDO::FETCH_ASSOC);

        if (!$product || $product['stock'] < $quantity) {
            // 库存不足，标记请求为失败
            $stmt = $pdo->prepare("UPDATE order_queue SET status = 'failed', processed_at = CURRENT_TIMESTAMP WHERE id = ?");
            $stmt->execute([$queueItem['id']]);
            $pdo->commit();
            // 这里可以给用户发站内信或邮件告知下单失败
            continue;
        }

        // 扣减库存
        $stmt = $pdo->prepare("UPDATE products SET stock = stock - ? WHERE id = ?");
        $stmt->execute([$quantity, $productId]);

        // 创建正式订单
        $stmt = $pdo->prepare("INSERT INTO orders (user_id, product_id, quantity, status) VALUES (?, ?, ?, 'confirmed')");
        $stmt->execute([$userId, $productId, $quantity]);

        // 标记请求处理成功
        $stmt = $pdo->prepare("UPDATE order_queue SET status = 'success', processed_at = CURRENT_TIMESTAMP WHERE id = ?");
        $stmt->execute([$queueItem['id']]);

        $pdo->commit();
        // 给用户发通知告知下单成功

    } catch(PDOException $e) {
        $pdo->rollBack();
        // 如果处理出错，把请求改回pending，方便下次重试
        if (isset($queueItem['id'])) {
            $stmt = $pdo->prepare("UPDATE order_queue SET status = 'pending' WHERE id = ?");
            $stmt->execute([$queueItem['id']]);
        }
        // 记录错误日志
        error_log("队列处理出错：" . $e->getMessage());
        sleep(2); // 出错后休眠2秒再重试
    }
}

4. 让脚本后台跑起来 #

Linux下执行这条命令即可：

nohup php /path/to/queue_processor.php >> /path/to/queue_log.log 2>&1 &

这样脚本会一直运行，每隔1秒检查一次队列，有请求就处理。

三、进阶：用Redis队列提升高并发场景性能 #

如果你的用户量很大，MySQL队列的性能可能跟不上，这时可以用Redis的List结构做队列，处理速度快很多：

1. 用户下单时把请求扔进Redis队列 #

// 连接Redis
$redis = new Redis();
$redis->connect('localhost', 6379);

// 把下单请求序列化为JSON存入队列
$requestData = json_encode([
    'user_id' => $_SESSION['user_id'],
    'product_id' => $_POST['product_id'],
    'quantity' => 1
]);
$redis->lPush('order_queue', $requestData);

echo "你的下单请求已提交，正在排队处理，请稍后查看订单状态~";

2. 队列处理脚本（常驻后台） #

$redis = new Redis();
$redis->connect('localhost', 6379);
$pdo = new PDO('mysql:host=localhost;dbname=your_db;charset=utf8mb4', 'db_user', 'db_pass');
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

while(true) {
    // 用阻塞式弹出队列尾部的请求，没有请求时会阻塞，不用轮询浪费资源
    $requestData = $redis->brPop('order_queue', 5); // 5秒超时
    if (!$requestData) {
        continue;
    }
    $request = json_decode($requestData[1], true);
    $userId = $request['user_id'];
    $productId = $request['product_id'];
    $quantity = $request['quantity'];

    // 下面的处理逻辑和MySQL队列的处理部分完全一样：
    // 开启事务→查库存（加锁）→扣库存→创建订单→更新状态
    // ... 这里省略重复代码，直接用之前的处理逻辑就行
}

同样用nohup命令让脚本后台运行即可，Redis的阻塞式弹出比MySQL轮询高效很多。

四、几个关键注意事项 #

• 用户体验要跟上：因为是异步处理，一定要给用户明确的反馈，比如“请求已提交，正在处理”，然后通过站内信、邮件或者页面刷新让用户知道结果。
• 重试机制不能少：如果处理请求时出错（比如数据库断连），一定要把请求放回队列，避免丢失用户的下单请求。
• 并发控制要做好：MySQL队列里用FOR UPDATE锁定请求，防止多个处理进程同时处理同一条；Redis队列因为是单线程取请求，天然不会有这个问题。
• 根据规模选方案：中小用户量用MySQL队列足够，不用额外加依赖；高并发场景直接上Redis队列，性能提升明显。

另外，其实还有数据库层面的解决方案（比如乐观锁、悲观锁），但你明确问的是用队列实现依次下单，所以重点讲了队列方案，如果需要了解其他方案也可以随时问~

内容的提问来源于stack exchange，提问作者Joede