IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

PHP 队列异步扣库存

wen PHP项目 3

本文目录导读：

PHP 队列异步扣库存

  1. 使用Redis队列实现
  2. 使用Laravel队列
  3. 数据库表结构
  4. 完整使用示例
  5. 性能优化建议

我来为你提供几种PHP队列异步扣库存的解决方案：

使用Redis队列实现

基础封装类

<?php
class RedisQueue {
    private $redis;
    public function __construct($config) {
        $this->redis = new Redis();
        $this->redis->connect($config['host'], $config['port']);
        if (isset($config['password'])) {
            $this->redis->auth($config['password']);
        }
        $this->redis->select($config['db'] ?? 0);
    }
    // 入队
    public function push($queueName, $data) {
        $json = json_encode($data);
        return $this->redis->lpush($queueName, $json);
    }
    // 出队（阻塞式）
    public function pop($queueName, $timeout = 0) {
        $result = $this->redis->brpop($queueName, $timeout);
        return $result ? json_decode($result[1], true) : false;
    }
    // 非阻塞出队
    public function popNonBlocking($queueName) {
        $result = $this->redis->rpop($queueName);
        return $result ? json_decode($result, true) : false;
    }
    // 获取队列长度
    public function length($queueName) {
        return $this->redis->llen($queueName);
    }
}
?>

库存服务类

<?php
class StockService {
    private $redis;
    private $db;
    public function __construct(PDO $db, RedisQueue $queue) {
        $this->db = $db;
        $this->queue = $queue;
    }
    // 扣减库存请求（入队）
    public function deductStockRequest($productId, $quantity, $orderId) {
        // 生成唯一请求ID
        $requestId = uniqid('stock_', true);
        $data = [
            'request_id' => $requestId,
            'product_id' => $productId,
            'quantity'   => $quantity,
            'order_id'   => $orderId,
            'timestamp'  => time()
        ];
        // 将扣库存请求加入队列
        $this->queue->push('stock_queue', $data);
        return [
            'success' => true,
            'request_id' => $requestId,
            'message' => '扣减请求已加入队列'
        ];
    }
    // 处理库存扣减（消费者）
    public function processStockDeduction($data) {
        try {
            // 开启事务
            $this->db->beginTransaction();
            // 加锁查询当前库存（使用悲观锁）
            $sql = "SELECT stock FROM products WHERE id = ? FOR UPDATE";
            $stmt = $this->db->prepare($sql);
            $stmt->execute([$data['product_id']]);
            $currentStock = $stmt->fetchColumn();
            if ($currentStock < $data['quantity']) {
                throw new Exception('库存不足');
            }
            // 扣减库存
            $sql = "UPDATE products SET stock = stock - ? WHERE id = ?";
            $stmt = $this->db->prepare($sql);
            $stmt->execute([$data['quantity'], $data['product_id']]);
            // 记录扣减日志
            $this->recordDeductLog($data);
            // 提交事务
            $this->db->commit();
            return [
                'success' => true,
                'request_id' => $data['request_id'],
                'message' => '库存扣减成功'
            ];
        } catch (Exception $e) {
            $this->db->rollBack();
            // 记录失败日志
            $this->recordFailLog($data, $e->getMessage());
            return [
                'success' => false,
                'request_id' => $data['request_id'],
                'message' => '库存扣减失败: ' . $e->getMessage()
            ];
        }
    }
    // 记录扣减日志
    private function recordDeductLog($data) {
        $sql = "INSERT INTO stock_logs 
                (request_id, product_id, quantity, order_id, status, created_at) 
                VALUES (?, ?, ?, ?, 'success', NOW())";
        $stmt = $this->db->prepare($sql);
        $stmt->execute([
            $data['request_id'],
            $data['product_id'],
            $data['quantity'],
            $data['order_id']
        ]);
    }
    // 记录失败日志
    private function recordFailLog($data, $error) {
        $sql = "INSERT INTO stock_logs 
                (request_id, product_id, quantity, order_id, status, error_log, created_at) 
                VALUES (?, ?, ?, ?, 'failed', ?, NOW())";
        $stmt = $this->db->prepare($sql);
        $stmt->execute([
            $data['request_id'],
            $data['product_id'],
            $data['quantity'],
            $data['order_id'],
            $error
        ]);
    }
}
?>

消费者进程

<?php
// consumer.php
require_once 'RedisQueue.php';
require_once 'StockService.php';
class StockConsumer {
    private $queue;
    private $stockService;
    private $running = true;
    public function __construct(StockService $stockService, RedisQueue $queue) {
        $this->queue = $queue;
        $this->stockService = $stockService;
    }
    public function run($processCount = 1) {
        // 多进程处理（使用pcntl扩展）
        if ($processCount > 1 && function_exists('pcntl_fork')) {
            $this->runMultiProcess($processCount);
        } else {
            $this->consumeLoop();
        }
    }
    // 单进程消费循环
    private function consumeLoop() {
        echo "开始消费库存队列...\n";
        while ($this->running) {
            try {
                // 从队列取出任务，阻塞超时30秒
                $task = $this->queue->pop('stock_queue', 30);
                if ($task) {
                    echo "处理任务: " . $task['request_id'] . "\n";
                    // 处理库存扣减
                    $result = $this->stockService->processStockDeduction($task);
                    echo "处理结果: " . json_encode($result) . "\n";
                    // 可以在这里记录处理结果到日志文件
                    error_log(json_encode($result) . "\n", 3, '/var/log/stock_consumer.log');
                }
                usleep(100000); // 100ms延迟防CPU过高
            } catch (Exception $e) {
                error_log("消费异常: " . $e->getMessage() . "\n", 3, '/var/log/stock_consumer_error.log');
                sleep(5);
            }
        }
    }
    // 多进程消费
    private function runMultiProcess($processCount) {
        echo "使用 {$processCount} 个进程消费...\n";
        for ($i = 0; $i < $processCount; $i++) {
            $pid = pcntl_fork();
            if ($pid == -1) {
                die("无法创建子进程");
            } elseif ($pid) {
                // 父进程继续创建其他子进程
                continue;
            } else {
                // 子进程消费
                $this->consumeLoop();
                exit(0);
            }
        }
        // 父进程等待所有子进程结束
        while (pcntl_waitpid(0, $status) != -1) {
            // 等待
        }
    }
    public function stop() {
        $this->running = false;
    }
}
// 使用示例
$dbConfig = [
    'host' => 'localhost',
    'dbname' => 'test',
    'user' => 'root',
    'pass' => 'password'
];
// 创建PDO连接
$dsn = "mysql:dbname={$dbConfig['dbname']};host={$dbConfig['host']};charset=utf8mb4";
$pdo = new PDO($dsn, $dbConfig['user'], $dbConfig['pass']);
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
// 创建队列实例
$redisConfig = [
    'host' => '127.0.0.1',
    'port' => 6379,
    'db' => 0
];
$queue = new RedisQueue($redisConfig);
// 创建库存服务
$stockService = new StockService($pdo, $queue);
// 启动消费者
$consumer = new StockConsumer($stockService, $queue);
$consumer->run(3); // 使用3个进程消费
?>

使用Laravel队列

创建任务

<?php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Redis;
class DeductStockJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    protected $productId;
    protected $quantity;
    protected $orderId;
    public function __construct($productId, $quantity, $orderId)
    {
        $this->productId = $productId;
        $this->quantity = $quantity;
        $this->orderId = $orderId;
    }
    public function handle()
    {
        // 使用Laravel自带的Redis锁
        $lock = Redis::lock('stock_' . $this->productId, 10);
        try {
            if ($lock->get()) {
                DB::transaction(function () {
                    // 查询库存
                    $product = DB::table('products')
                        ->where('id', $this->productId)
                        ->lockForUpdate()
                        ->first();
                    if ($product->stock < $this->quantity) {
                        throw new \Exception('库存不足');
                    }
                    // 扣减库存
                    DB::table('products')
                        ->where('id', $this->productId)
                        ->decrement('stock', $this->quantity);
                    // 记录日志
                    DB::table('stock_logs')->insert([
                        'product_id' => $this->productId,
                        'quantity' => $this->quantity,
                        'order_id' => $this->orderId,
                        'status' => 'success',
                        'created_at' => now()
                    ]);
                    // 更新Redis缓存
                    Redis::hincrby('product_stock', $this->productId, -$this->quantity);
                });
            }
        } finally {
            $lock->release();
        }
    }
    // 失败处理
    public function failed($exception)
    {
        // 记录失败日志
        \Log::error('库存扣减失败', [
            'product_id' => $this->productId,
            'order_id' => $this->orderId,
            'error' => $exception->getMessage()
        ]);
        // 可以通知用户或进行补偿操作
    }
}
// 调用方式
DeductStockJob::dispatch($productId, $quantity, $orderId)->onQueue('stock');
?>

数据库表结构

-- 商品表
CREATE TABLE products (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(255) NOT NULL,
    stock INT NOT NULL DEFAULT 0,
    version INT NOT NULL DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
-- 库存日志表
CREATE TABLE stock_logs (
    id INT PRIMARY KEY AUTO_INCREMENT,
    request_id VARCHAR(50) UNIQUE,
    product_id INT NOT NULL,
    quantity INT NOT NULL,
    order_id VARCHAR(50) NOT NULL,
    status ENUM('success', 'failed', 'pending') DEFAULT 'pending',
    error_log TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    processed_at TIMESTAMP NULL,
    INDEX idx_product_status (product_id, status),
    INDEX idx_order_id (order_id)
);
-- 订单表
CREATE TABLE orders (
    id INT PRIMARY KEY AUTO_INCREMENT,
    order_no VARCHAR(50) UNIQUE,
    status ENUM('pending', 'paid', 'cancelled') DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 订单商品表
CREATE TABLE order_items (
    id INT PRIMARY KEY AUTO_INCREMENT,
    order_id INT NOT NULL,
    product_id INT NOT NULL,
    quantity INT NOT NULL,
    price DECIMAL(10,2) NOT NULL,
    INDEX idx_order_product (order_id, product_id)
);

完整使用示例

<?php
// 下单接口示例
class OrderController {
    public function createOrder($request) {
        try {
            // 创建订单
            $orderId = $this->createOrderRecord();
            // 扣减库存请求（异步）
            $stockService->deductStockRequest(
                $request['product_id'],
                $request['quantity'],
                $orderId
            );
            // 返回订单信息
            return response()->json([
                'success' => true,
                'order_id' => $orderId,
                'message' => '订单创建成功',
                'include_async' => true // 标记扣库存是异步的
            ]);
        } catch (Exception $e) {
            return response()->json([
                'success' => false,
                'message' => $e->getMessage()
            ], 500);
        }
    }
}
?>

性能优化建议

Redis哨兵/集群配置

// 使用Redis集群时
$options = [
    'cluster' => true,
    'prefix' => 'stock_',
];
$redis = new RedisCluster(NULL, ['127.0.0.1:7000', '127.0.0.1:7001']);

幂等性处理

// 添加幂等性检查
public function checkIdempotent($requestId, $productId, $orderId) {
    $sql = "SELECT COUNT(*) FROM stock_logs 
            WHERE request_id = ? OR (product_id = ? AND order_id = ?)";
    $stmt = $this->db->prepare($sql);
    $stmt->execute([$requestId, $productId, $orderId]);
    return $stmt->fetchColumn() > 0;
}

监控和告警

// 队列长度监控
$queueLength = $this->queue->length('stock_queue');
if ($queueLength > 1000) {
    // 发送告警
    $this->sendAlert('库存队列堆积，当前长度: ' . $queueLength);
}

这套方案实现了：

  • 异步处理库存扣减
  • 防止超卖（使用数据库事务和锁）
  • 失败重试机制
  • 操作日志记录
  • 性能较优（批量处理能力强）

上一篇PHP Redis 原子扣减

下一篇PHP 并发扣减库存方案

相关文章

抱歉，评论功能暂时关闭!