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

PHP 队列异步扣库存

wen PHP项目 215

本文目录导读：

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);
}

这套方案实现了：

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

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇PHP Redis 原子扣减

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

相关文章

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

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝