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, // 重试次数
];
}
}
- 性能优化：利用 Redis 原子操作（INCR/DECR 或 Lua脚本）实现超高并发扣减
- 数据一致性：最终通过数据库事务保证数据一致性
- 降级策略：Redis故障时可降级到数据库锁方案
- 异步处理：大量并发请求可通过消息队列削峰填谷
- 超卖防护：所有方案都需确保库存非负
实际流程建议
- 简单场景：使用 Redis DECR 原子操作
- 电商场景：Redis 缓存 + 数据库兜底 + 异步同步
- 高并发秒杀：消息队列 + 多级缓存 + 数据库最终一致性
