PHP + Redis实现订单超时

ALTER TABLE orders ADD COLUMN status TINYINT DEFAULT 0 COMMENT '0-待支付 1-已支付 2-已取消 3-已超时';
ALTER TABLE orders ADD COLUMN expire_time DATETIME COMMENT '订单超时时间';

$orderId = 123456;  // 假设已生成
$expireSeconds = 1800; // 30分钟超时
$expireTimestamp = time() + $expireSeconds;

// 将订单ID加入超时监控队列，score为超时时间戳
$redis->zAdd('order:timeout', $expireTimestamp, $orderId);

<?php
// timeout_consumer.php 用 Supervisor 管理
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$pdo = new PDO('mysql:host=localhost;dbname=test', 'root', 'password');

while (true) {
    $now = time();
    // 获取所有超时的订单ID（score <= 当前时间戳）
    // ZRANGEBYSCORE key -inf (now 按顺序取出，每次最多取100个
    $timeoutOrderIds = $redis->zRangeByScore('order:timeout', '-inf', $now, ['limit' => [0, 100]]);
    
    foreach ($timeoutOrderIds as $orderId) {
        // 开启事务，避免并发问题
        $pdo->beginTransaction();
        try {
            // 使用悲观锁锁定订单行，防止重复处理
            $stmt = $pdo->prepare("SELECT status FROM orders WHERE id = ? FOR UPDATE");
            $stmt->execute([$orderId]);
            $order = $stmt->fetch();
            
            if ($order && $order['status'] == 0) { // 仅处理待支付订单
                // 更新订单状态为超时取消
                $pdo->prepare("UPDATE orders SET status = 3 WHERE id = ?")->execute([$orderId]);
                
                // 【关键】恢复库存：根据订单的商品ID，增加Redis中的秒杀库存
                $goodsId = $pdo->prepare("SELECT goods_id FROM orders WHERE id = ?")->execute([$orderId]);
                $redis->incr("seckill:stock:$goodsId");
                // 注意：也要移除用户抢购记录，让该用户可以再次抢购
                $redis->sRem("seckill:users:$goodsId", $userId);
                
                $pdo->commit();
                
                // 从超时队列中移除
                $redis->zRem('order:timeout', $orderId);
            } else {
                $pdo->rollBack();
                // 订单已被处理，直接移除队列记录
                $redis->zRem('order:timeout', $orderId);
            }
        } catch (Exception $e) {
            $pdo->rollBack();
            // 记录错误日志，稍后重试（不移除队列，下次扫描还会处理）
        }
    }
    
    // 避免空转消耗CPU，没有任务时休眠1秒
    if (empty($timeoutOrderIds)) {
        sleep(1);
    }
}

// 支付成功回调
function paySuccess($orderId) {
    $redis->zRem('order:timeout', $orderId);
    // 更新订单状态为已支付...
}

$shard = $orderId % 10;
$redis->zAdd("order:timeout:$shard", $expireTimestamp, $orderId);

-- 查找超过30分钟未支付的订单
UPDATE orders o 
JOIN goods g ON o.goods_id = g.id
SET o.status = 3, g.seckill_stock = g.seckill_stock + 1
WHERE o.status = 0 AND o.expire_time < NOW();

用户秒杀成功
    ↓
订单写入 MySQL（status=0, expire_time=now+30min）
    ↓
redis.zAdd('order:timeout', expire_timestamp, order_id)
    ↓
[常驻进程] 每1秒扫描 zRangeByScore（-inf, now）
    ↓
发现超时订单
    ↓
MySQL事务：UPDATE orders SET status=3 WHERE id=? AND status=0
    ↓
redis.incr('seckill:stock:goods_id')  # 恢复库存
    ↓
redis.sRem('seckill:users:goods_id', user_id)  # 移除用户记录
    ↓
redis.zRem('order:timeout', order_id)
    ↓
提交事务

// 订单创建时：添加 Stream 消息
$streamKey = "order:stream";
$orderId = 123456;
$expireAt = time() + 1800;
$redis->xAdd($streamKey, '*', [
    'order_id' => $orderId,
    'expire_at' => $expireAt
]);

// 消费者进程（可启动多个）
$groupName = "order_timeout_group";
$consumerName = "consumer_1";
$redis->xGroup('CREATE', $streamKey, $groupName, 0, true); // 创建消费者组

while (true) {
    // 阻塞读取新消息，超时1秒
    $messages = $redis->xReadGroup($groupName, $consumerName, [$streamKey => '>'], 1, 1000);
    foreach ($messages[$streamKey] as $msgId => $data) {
        $expireAt = $data['expire_at'];
        $now = time();
        if ($now < $expireAt) {
            // 未到超时时间，等待
            $sleep = $expireAt - $now;
            usleep($sleep * 1000000);
        }
        // 执行超时取消逻辑（加锁、更新订单、恢复库存）
        $this->cancelOrder($data['order_id']);
        $redis->xAck($streamKey, $groupName, $msgId);
    }
}

// 订单创建时，计算桶位置（假设每10秒一个桶）
$bucketInterval = 10; // 秒
$bucketIndex = floor(time() / $bucketInterval);
$redis->sAdd("order🪣$bucketIndex", $orderId);

// 独立的定时器进程（每10秒触发一次）
$currentBucket = floor(time() / $bucketInterval);
while (true) {
    $bucketKey = "order🪣$currentBucket";
    $orderIds = $redis->sMembers($bucketKey);
    foreach ($orderIds as $orderId) {
        // 检查订单是否超时（实际时间可能已过桶边界）
        if (订单未支付) {
            取消订单 + 恢复库存;
        }
    }
    $redis->del($bucketKey);
    sleep($bucketInterval);
    $currentBucket++;
}

// 订阅脚本（PHP 使用 Redis 的 subscribe）
$redis->subscribe(['__keyevent@0__:expired'], function ($redis, $channel, $message) {
    if (strpos($message, 'order:timeout:') === 0) {
        $orderId = substr($message, strlen('order:timeout:'));
        // 调用取消逻辑（需保证幂等）
        cancelOrder($orderId);
    }
});