PHP队列处理秒杀订单

wen PHP项目 2026-08-11 125

本文目录导读：

PHP队列处理秒杀订单

Redis队列方案（推荐）
高级优化方案
最终订单处理
监控和优化
配置示例
最佳实践建议

我来为你提供一个完整的PHP秒杀系统队列处理方案,包含高并发场景下的最佳实践。

Redis队列方案（推荐）

生产者：秒杀请求入队

<?php
class SeckillProducer {
    private $redis;
    public function __construct() {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    /**
     * 秒杀请求入队
     */
    public function enqueueSeckill($userId, $productId) {
        // 使用Lua脚本保证原子性
        $lua = <<<LUA
            -- 检查是否重复秒杀
            local user_key = KEYS[1] .. ARGV[1]
            if redis.call('exists', user_key) == 1 then
                return 0
            end
            -- 检查库存
            local stock = redis.call('get', KEYS[2])
            if not stock or tonumber(stock) <= 0 then
                return 0
            end
            -- 扣减库存
            redis.call('dec', KEYS[2])
            -- 记录用户
            redis.call('set', user_key, 1, 'ex', 86400)
            -- 入队
            local queue_key = KEYS[3]
            redis.call('lpush', queue_key, ARGV[1] .. ':' .. ARGV[2])
            return 1
LUA;
        $result = $this->redis->eval(
            $lua,
            ['user:' . $productId, 'stock:' . $productId, 'seckill_queue:' . $productId],
            [$userId, $productId]
        );
        return $result === 1;
    }
}

消费者：异步处理订单

<?php
class SeckillConsumer {
    private $redis;
    private $pdo;
    public function __construct() {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
        // 数据库连接
        $this->pdo = new PDO(
            'mysql:host=localhost;dbname=seckill',
            'username',
            'password'
        );
    }
    /**
     * 处理队列中的订单
     */
    public function consume() {
        set_time_limit(0);
        while (true) {
            // 从队尾取出请求
            $data = $this->redis->rpop('seckill_queue:product_' . PRODUCT_ID);
            if ($data) {
                list($userId, $productId) = explode(':', $data);
                try {
                    // 创建订单
                    $this->pdo->beginTransaction();
                    $sql = "INSERT INTO orders (user_id, product_id, status, created_at) 
                            VALUES (?, ?, 'processing', NOW())";
                    $stmt = $this->pdo->prepare($sql);
                    $stmt->execute([$userId, $productId]);
                    $this->pdo->commit();
                    // 发送通知消息（如通过消息队列）
                    $this->notifyUser($userId, $productId, '秒杀成功');
                } catch (Exception $e) {
                    $this->pdo->rollBack();
                    // 库存回补
                    $this->redis->incr('stock:' . $productId);
                    // 记录失败
                    $this->logFailure($userId, $productId, $e->getMessage());
                }
            } else {
                // 队列为空，等待
                usleep(100000); // 100ms
            }
        }
    }
}

高级优化方案

使用延时队列处理超时订单

<?php
class SeckillTimeoutHandler {
    private $redis;
    public function __construct() {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    /**
     * 将订单加入延时队列
     */
    public function addToDelayQueue($orderId, $timeoutSeconds = 300) {
        $this->redis->zAdd('timeout_orders', time() + $timeoutSeconds, $orderId);
    }
    /**
     * 检查超时订单
     */
    public function handleTimeoutOrders() {
        $now = time();
        // 获取所有超时的订单
        $timeoutOrders = $this->redis->zRangeByScore('timeout_orders', 0, $now);
        foreach ($timeoutOrders as $orderId) {
            // 更新订单状态为超时
            $this->updateOrderStatus($orderId, 'timeout');
            // 库存回补
            $this->restoreStock($orderId);
            // 从延时队列移除
            $this->redis->zRem('timeout_orders', $orderId);
        }
    }
}

使用消息队列（RabbitMQ）

<?php
class RabbitMQSeckill {
    private $connection;
    private $channel;
    public function __construct() {
        $this->connection = new AMQPConnection([
            'host' => 'localhost',
            'port' => 5672,
            'login' => 'guest',
            'password' => 'guest'
        ]);
        $this->channel = $this->connection->channel();
        // 声明交换机
        $this->channel->exchange_declare(
            'seckill_exchange', 
            'direct', 
            false, 
            true
        );
        // 声明队列
        $this->channel->queue_declare(
            'seckill_queue',
            false,
            true,
            false,
            false,
            false,
            ['x-message-ttl' => 300000] // 5分钟过期
        );
        $this->channel->queue_bind('seckill_queue', 'seckill_exchange', 'seckill');
    }
    /**
     * 发送秒杀请求
     */
    public function sendSeckillRequest($userId, $productId) {
        $message = new AMQPMessage(json_encode([
            'user_id' => $userId,
            'product_id' => $productId,
            'timestamp' => time()
        ]), [
            'delivery_mode' => 2, // 持久化
            'content_type' => 'application/json'
        ]);
        $this->channel->basic_publish(
            $message,
            'seckill_exchange',
            'seckill'
        );
    }
    /**
     * 消费者
     */
    public function consume() {
        $callback = function($msg) {
            $data = json_decode($msg->body, true);
            // 处理订单
            $this->processOrder($data['user_id'], $data['product_id']);
            // 确认消息
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };
        $this->channel->basic_qos(null, 10, null); // 预取10条
        $this->channel->basic_consume(
            'seckill_queue',
            '',
            false,
            false,
            false,
            false,
            $callback
        );
        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }
}

最终订单处理

<?php
class FinalOrderProcessor {
    /**
     * 批量处理订单（优化性能）
     */
    public function batchProcess($batchSize = 100) {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        $pdo = new PDO('mysql:host=localhost;dbname=seckill', 'user', 'pass');
        $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
        // 使用管道批量处理
        $redis->multi();
        for ($i = 0; $i < $batchSize; $i++) {
            $redis->rPop('seckill_queue');
        }
        $results = $redis->exec();
        // 批量插入数据库
        $sql = "INSERT INTO orders (user_id, product_id, status) VALUES ";
        $params = [];
        $placeholders = [];
        foreach ($results as $data) {
            if ($data) {
                list($userId, $productId) = explode(':', $data);
                $placeholders[] = "(?, ?, 'pending')";
                $params[] = $userId;
                $params[] = $productId;
            }
        }
        if (!empty($placeholders)) {
            $sql .= implode(',', $placeholders);
            $stmt = $pdo->prepare($sql);
            $stmt->execute($params);
        }
    }
    /**
     * 使用信号量控制并发
     */
    public function processWithSemaphore($userId, $productId) {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        // 获取分布式锁
        $lockKey = "lock:seckill:$productId";
        $lockValue = uniqid();
        // 尝试获取锁
        if ($redis->set($lockKey, $lockValue, ['NX', 'EX' => 5])) {
            try {
                // 处理订单
                $this->createOrder($userId, $productId);
            } finally {
                // 释放锁
                if ($redis->get($lockKey) === $lockValue) {
                    $redis->del($lockKey);
                }
            }
        }
    }
}

监控和优化

<?php
class SeckillMonitor {
    /**
     * 监控队列状态
     */
    public function getQueueStatus() {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        return [
            'queue_length' => $redis->lLen('seckill_queue'),
            'stock' => $redis->get('stock:' . PRODUCT_ID),
            'success_count' => $redis->get('seckill:success_count'),
            'fail_count' => $redis->get('seckill:fail_count'),
        ];
    }
    /**
     * 告警阈值设置
     */
    public function checkThresholds() {
        $status = $this->getQueueStatus();
        // 队列积压告警
        if ($status['queue_length'] > 1000) {
            $this->sendAlert('Queue backlog high: ' . $status['queue_length']);
        }
        // 库存不足告警
        if ($status['stock'] < 10) {
            $this->sendAlert('Stock low: ' . $status['stock']);
        }
    }
}

配置示例

; php.ini 配置建议
max_execution_time = 0
memory_limit = 512M
; Redis 配置
redis.connection_pool_size = 100
; 数据库连接池
pdo.pool_size = 50

最佳实践建议

库存预热：秒杀前将库存加载到Redis
并发控制：使用Redis分布式锁控制库存扣减
异步处理：将订单创建放入队列，避免阻塞
死信队列：处理失败消息
幂等设计：防止重复下单
监控告警：实时监控系统状态

这个方案可以支持高并发秒杀场景,建议根据实际需求调整队列和缓存策略。

本文地址： https://bmjup.com/post/43655.html