PHP 并发扣减库存方案

wen PHP项目 2026-08-11 147

PHP 并发扣减库存方案

数据库乐观锁方案

// 使用 version 字段实现乐观锁
public function deductStock($productId, $quantity) {
    $pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'pass');
    $pdo->beginTransaction();
    try {
        // 1. 查询当前库存和版本号
        $stmt = $pdo->prepare('SELECT stock, version FROM products WHERE id = ? FOR UPDATE');
        $stmt->execute([$productId]);
        $product = $stmt->fetch(PDO::FETCH_ASSOC);
        // 2. 检查库存是否充足
        if ($product['stock'] < $quantity) {
            throw new Exception('库存不足');
        }
        // 3. 更新库存并使用版本号控制并发
        $sql = 'UPDATE products 
                SET stock = stock - :quantity,
                    version = version + 1
                WHERE id = :id AND version = :version';
        $stmt = $pdo->prepare($sql);
        $stmt->execute([
            ':quantity' => $quantity,
            ':id' => $productId,
            ':version' => $product['version']
        ]);
        // 4. 检查影响行数
        if ($stmt->rowCount() === 0) {
            throw new Exception('并发冲突，请重试');
        }
        $pdo->commit();
        return true;
    } catch (Exception $e) {
        $pdo->rollBack();
        throw $e;
    }
}

Redis 原子操作方案

class StockManager {
    private $redis;
    private $lockKey = 'stock:lock';
    public function __construct() {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    // 方案A：使用 Redis 的 DECR 命令（最佳性能）
    public function deductStockRedis($productId, $quantity) {
        $stockKey = "stock:{$productId}";
        // 使用 Lua 脚本保证原子性
        $luaScript = `
            local stock = redis.call('get', KEYS[1])
            if not stock or tonumber(stock) < tonumber(ARGV[1]) then
                return 0
            end
            return redis.call('decrby', KEYS[1], ARGV[1])
        `;
        $result = $this->redis->eval($luaScript, [$stockKey, $quantity], 1);
        if ($result === false) {
            throw new Exception('扣减失败');
        }
        return $result;
    }
    // 方案B：使用分布式锁实现
    public function deductStockWithLock($productId, $quantity) {
        $lockKey = "lock:{$productId}";
        $lockValue = uniqid();
        // 获取锁
        $acquired = $this->redis->set(
            $lockKey, 
            $lockValue, 
            ['NX', 'EX' => 30]
        );
        if (!$acquired) {
            throw new Exception('系统繁忙，请重试');
        }
        try {
            // 检查并扣减库存
            $stockKey = "stock:{$productId}";
            $currentStock = $this->redis->get($stockKey);
            if ($currentStock === false || $currentStock < $quantity) {
                throw new Exception('库存不足');
            }
            $newStock = $this->redis->decrby($stockKey, $quantity);
            $this->logOperation($productId, $quantity, $newStock);
            return $newStock;
        } finally {
            // 释放锁
            $this->releaseLock($lockKey, $lockValue);
        }
    }
    private function releaseLock($lockKey, $lockValue) {
        // 使用 Lua 确保只有持有者才能释放锁
        $script = `
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            end
            return 0
        `;
        $this->redis->eval($script, [$lockKey, $lockValue], 1);
    }
    private function logOperation($productId, $quantity, $newStock) {
        // 记录操作日志
        error_log("Product {$productId} deducted {$quantity}, new stock: {$newStock}");
    }
}

消息队列方案

// 使用 RabbitMQ 或 Redis Queue 异步处理
class StockQueueProcessor {
    public function __construct() {
        // 初始化队列连接
        $this->queue = new RedisQueue('stock_queue');
    }
    // 加入扣减请求到队列
    public function enqueueDeductRequest($productId, $quantity, $orderId) {
        $data = [
            'product_id' => $productId,
            'quantity' => $quantity,
            'order_id' => $orderId,
            'timestamp' => time()
        ];
        return $this->queue->push(json_encode($data));
    }
    // 消费者处理队列
    public function processQueue() {
        while ($item = $this->queue->pop()) {
            $request = json_decode($item, true);
            try {
                // 调用库存扣减服务
                $this->deductStock($request['product_id'], $request['quantity']);
                // 成功后通知订单服务
                $this->notifyOrderService($request['order_id'], true);
            } catch (Exception $e) {
                // 失败则重新入队或标记失败
                $this->queue->push($item); // 稍后重试
                $this->notifyOrderService($request['order_id'], false);
            }
        }
    }
    private function notifyOrderService($orderId, $success) {
        // 通知订单服务更新状态
    }
}

多级缓存方案

class MultiLevelStockManager {
    public function deductWithCache($productId, $quantity) {
        // 1. 先检查 Redis 缓存
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        $stockKey = "stock:{$productId}";
        $localStock = $redis->get($stockKey);
        if ($localStock !== false) {
            if (($localStock = $redis->decrby($stockKey, $quantity)) >= 0) {
                // 扣减成功，异步更新数据库
                $this->asyncUpdateDB($productId, $quantity);
                return true;
            } else {
                // 库存不足
                $redis->set($stockKey, 0);
                return false;
            }
        }
        // 2. 缓存不存在，从数据库加载
        return $this->loadFromDB($productId, $quantity);
    }
    private function loadFromDB($productId, $quantity) {
        $pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'pass');
        try {
            $pdo->beginTransaction();
            // 加锁查询
            $stmt = $pdo->prepare('SELECT stock FROM products WHERE id = ? FOR UPDATE');
            $stmt->execute([$productId]);
            $stock = $stmt->fetchColumn();
            if ($stock < $quantity) {
                $pdo->rollBack();
                return false;
            }
            // 更新库存
            $updateStmt = $pdo->prepare('UPDATE products SET stock = stock - ? WHERE id = ?');
            $updateStmt->execute([$quantity, $productId]);
            $pdo->commit();
            // 更新 Redis 缓存
            $this->updateRedisCache($productId);
            return true;
        } catch (Exception $e) {
            $pdo->rollBack();
            throw $e;
        }
    }
}

最佳实践建议

class StockService {
    public function qaTest() {
        return "开始测试方案";
    }
    public static function getConfig() {
        return [
            'strategy' => 'hybrid', // 混合模式
            'single_op' => 'redis', // 单次操作使用Redis
            'batch_op' => 'queue', // 批量操作使用消息队列 
            'fallback' => 'database', // 兜底方案
            'timeout' => 3000, // 超时时间
            'retry' => 3, // 重试次数
        ];
    }
}