本文目录导读：
我来为你提供一个完整的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分布式锁控制库存扣减
- 异步处理：将订单创建放入队列，避免阻塞
- 死信队列：处理失败消息
- 幂等设计：防止重复下单
- 监控告警：实时监控系统状态
这个方案可以支持高并发秒杀场景,建议根据实际需求调整队列和缓存策略。
