Redis 延时队列详解

# ZSet 天然适合：
ZADD delay_queue 1718000000 "task_001"   # score=到期时间戳
ZADD delay_queue 1718000300 "task_002"   # 自动按时间排序

生产者                     Redis ZSet                      消费者(定时任务)
──────                    ──────────                      ──────────────
XADD delay_q              score  = 到期时间戳
score=到期时间            member = 任务数据
                          ┌─────────────────┐
                          │ 1718000000 task1 │  ← 最早到期
                          │ 1718000300 task2 │
                          │ 1718000600 task3 │
                          └─────────────────┘
                                  ↓
                          ZRANGEBYSCORE 0 当前时间
                          取到 task1（已到期）
                                  ↓
                          执行 task1 → ZREM 删除

// 生产者：投递延时任务
$redis->zAdd('delay:orders', time() + 1800, json_encode([
    'order_id' => 12345,
    'action'   => 'auto_cancel',
]));
// 消费者：每秒轮询到期任务（有 BUG 的版本）
$now = time();
$tasks = $redis->zRangeByScore('delay:orders', 0, $now, ['limit' => [0, 1]]);
if ($tasks) {
    $task = $tasks[0];
    // ⚠️ 这里有 BUG！如果同时多个消费者拿到同一条
    $redis->zRem('delay:orders', $task);
    processTask($task);
}

-- 原子操作：查出到期任务 + 立刻删除 + 返回
local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, 1)
if #tasks == 0 then
    return nil
end
local task = tasks[1]
local removed = redis.call('ZREM', KEYS[1], task)
if removed == 1 then
    return task       -- 删除成功，返回任务
else
    return nil        -- 被别的消费者抢了
end

$lua = <<<'LUA'
local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, 1)
if #tasks == 0 then return nil end
local task = tasks[1]
if redis.call('ZREM', KEYS[1], task) == 1 then
    return task
else
    return nil
end
LUA;
// 定时任务循环执行
while (true) {
    $task = $redis->eval($lua, ['delay:orders', time()], 1);
    //                                      ↑ KEYS 部分       ↑ key数量
    if ($task) {
        $data = json_decode($task, true);
        echo "处理任务: {$data['order_id']}\n";
        processTask($data);
    } else {
        sleep(1);  // 没任务就等一下
    }
}

<?php
// ====== 生产者（下单时） ======
function createOrder($orderId) {
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6380);
    // 1. 创建订单...
    // 2. 投递延时任务：30分钟后自动取消
    $delayAt = time() + 1800;  // 30分钟
    $task = json_encode([
        'order_id' => $orderId,
        'action'   => 'auto_cancel',
        'create_at'=> date('Y-m-d H:i:s'),
    ]);
    $redis->zAdd('delay:orders', $delayAt, $task);
    echo "订单 {$orderId} 已创建，30分钟后未支付将自动取消\n";
}
// ====== 消费者（定时脚本） ======
$lua = <<<'LUA'
local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, 1)
if #tasks == 0 then return nil end
if redis.call('ZREM', KEYS[1], tasks[1]) == 1 then
    return tasks[1]
end
return nil
LUA;
while (true) {
    $task = $redis->eval($lua, ['delay:orders', time()], 1);
    if ($task) {
        $data = json_decode($task, true);
        // 检查订单是否已支付
        $order = getOrder($data['order_id']);
        if ($order['status'] === 'unpaid') {
            cancelOrder($data['order_id']);
            echo "⏰ 订单 {$data['order_id']} 超时未支付，已自动取消\n";
        } else {
            echo "✓ 订单 {$data['order_id']} 已支付，跳过\n";
        }
    } else {
        sleep(1);
    }
}

ZADD key score member
          ↑
     你自己指定的

ZADD delay_queue 1718000000 "task_001"
#                 ↑    时间戳就是 score，你自己算的
#                 score 决定了 ZSet 里的排序