PHP项目多级延时如何拆分不同时长任务队列

本文目录导读：

1. 核心架构设计
2. 多级时间轮实现
3. 基于Redis的延时队列实现
4. 多级队列分配器
5. 消费者进程实现
6. 监控与管理API
7. 使用示例

针对PHP项目的多级延时任务队列拆分,我提供一个完整的架构设计方案：

核心架构设计

时间轮 + 多级队列方案

<?php
// 延时任务队列管理器
class DelayedTaskManager 
{
    // 时间轮层级配置（单位：秒）
    private static $wheelConfig = [
        'seconds' => ['capacity' => 60, 'tick' => 1],      // 1秒精度，60个槽
        'minutes' => ['capacity' => 60, 'tick' => 60],     // 1分钟精度，60个槽
        'hours'   => ['capacity' => 24, 'tick' => 3600],   // 1小时精度，24个槽
        'days'    => ['capacity' => 30, 'tick' => 86400],  // 1天精度，30个槽
    ];
    private $wheels = [];
    private $currentTime;
    public function __construct() 
    {
        $this->currentTime = time();
        foreach (self::$wheelConfig as $level => $config) {
            $this->wheels[$level] = new TimeWheel($level, $config);
        }
    }
    // 添加延时任务
    public function addTask(Task $task, int $delay) 
    {
        $level = $this->determineLevel($delay);
        $this->wheels[$level]->addTask($task, $delay);
    }
    // 确定任务所属层级
    private function determineLevel(int $delay): string 
    {
        if ($delay <= 60) return 'seconds';        // 1分钟内
        if ($delay <= 3600) return 'minutes';       // 1小时内
        if ($delay <= 86400) return 'hours';        // 1天内
        return 'days';                              // 1天以上
    }
}

多级时间轮实现

<?php
class TimeWheel 
{
    private $level;
    private $capacity;    // 槽数量
    private $tick;        // 每个槽的时间间隔
    private $currentSlot = 0;
    private $slots = [];
    private $redis;
    public function __construct(string $level, array $config) 
    {
        $this->level = $level;
        $this->capacity = $config['capacity'];
        $this->tick = $config['tick'];
        $this->redis = new Redis();
        $this->initSlots();
    }
    // 初始化槽
    private function initSlots() 
    {
        for ($i = 0; $i < $this->capacity; $i++) {
            $this->slots[$i] = new Slot($this->redis, $this->getSlotKey($i));
        }
    }
    // 添加任务到指定槽
    public function addTask(Task $task, int $delay) 
    {
        $slotIndex = $this->calculateSlotIndex($delay);
        $this->slots[$slotIndex]->push($task);
    }
    // 计算目标槽索引
    private function calculateSlotIndex(int $delay): int 
    {
        $totalTicks = ceil($delay / $this->tick);
        return ($this->currentSlot + $totalTicks) % $this->capacity;
    }
    // 执行当前槽任务，并向下一级传递过期任务
    public function tick() 
    {
        $slot = $this->slots[$this->currentSlot];
        $tasks = $slot->popAll();
        foreach ($tasks as $task) {
            if ($task->getRemainingTime() <= 0) {
                // 立即执行
                $this->executeTask($task);
            } else {
                // 降级到更细粒度的时间轮
                $this->downgradeTask($task);
            }
        }
        $this->currentSlot = ($this->currentSlot + 1) % $this->capacity;
    }
    // 降级任务到更小层级
    private function downgradeTask(Task $task) 
    {
        $remainingTime = $task->getRemainingTime();
        $level = $this->determineLevel($remainingTime);
        if ($level !== $this->level) {
            // 转移到更细粒度的轮子
            global $taskManager;
            $taskManager->addTask($task, $remainingTime);
        } else {
            // 重新计算槽位
            $this->addTask($task, $remainingTime);
        }
    }
    private function getSlotKey($index) 
    {
        return "timewheel:{$this->level}:{$index}";
    }
}

基于Redis的延时队列实现

<?php
class RedisDelayedQueue 
{
    private $redis;
    private $zsetKey = 'delayed_tasks';
    private $listPrefix = 'task_list:';
    public function __construct() 
    {
        $this->redis = new Redis();
    }
    // 添加延时任务
    public function addTask(Task $task, int $delay): bool 
    {
        $score = microtime(true) + $delay;
        $taskId = $task->getId();
        // 使用Redis事务保证原子性
        $this->redis->multi();
        $this->redis->zAdd($this->zsetKey, $score, $taskId);
        $this->redis->hSet($this->listPrefix, $taskId, serialize($task));
        $this->redis->exec();
        return true;
    }
    // 获取到期任务
    public function getExpiredTasks(): array 
    {
        $currentTime = microtime(true);
        $tasks = [];
        // 获取到期的任务ID
        $expiredIds = $this->redis->zRangeByScore(
            $this->zsetKey, 
            0, 
            $currentTime
        );
        if (empty($expiredIds)) {
            return [];
        }
        // 获取任务内容并删除
        $this->redis->multi();
        foreach ($expiredIds as $id) {
            $taskData = $this->redis->hGet($this->listPrefix, $id);
            if ($taskData) {
                $tasks[$id] = unserialize($taskData);
                $this->redis->hDel($this->listPrefix, $id);
            }
        }
        $this->redis->zRem($this->zsetKey, ...$expiredIds);
        $this->redis->exec();
        return $tasks;
    }
    // 批量获取到期任务（分页）
    public function getExpiredTasksByPage(int $limit = 100): array 
    {
        $currentTime = microtime(true);
        // 使用ZRANGEBYSCORE with LIMIT
        $expiredIds = $this->redis->zRangeByScore(
            $this->zsetKey, 
            0, 
            $currentTime, 
            ['limit' => [0, $limit]]
        );
        if (empty($expiredIds)) {
            return [];
        }
        $tasks = [];
        $this->redis->multi();
        foreach ($expiredIds as $id) {
            $taskData = $this->redis->hGet($this->listPrefix, $id);
            if ($taskData) {
                $tasks[$id] = unserialize($taskData);
                $this->redis->hDel($this->listPrefix, $id);
            }
        }
        $this->redis->zRem($this->zsetKey, ...$expiredIds);
        $this->redis->exec();
        return $tasks;
    }
}

多级队列分配器

<?php
class QueueDispatcher 
{
    // 队列级别配置
    private $queueLevels = [
        'immediate' => [
            'driver' => 'redis_list',
            'max_delay' => 5,        // 5秒内
            'concurrency' => 10,
            'retry_times' => 3
        ],
        'short' => [
            'driver' => 'redis_zset',
            'max_delay' => 300,      // 5分钟内
            'concurrency' => 5,
            'retry_times' => 3
        ],
        'medium' => [
            'driver' => 'mysql',
            'max_delay' => 3600,     // 1小时内
            'concurrency' => 2,
            'retry_times' => 5
        ],
        'long' => [
            'driver' => 'file',
            'max_delay' => 86400 * 7, // 7天内
            'concurrency' => 1,
            'retry_times' => 10
        ]
    ];
    private $queues = [];
    private $logger;
    public function __construct() 
    {
        $this->logger = new Logger();
        $this->initQueues();
    }
    // 分配任务到合适的队列
    public function dispatch(Task $task): bool 
    {
        $delay = $task->getDelay();
        $level = $this->determineQueueLevel($delay);
        try {
            $queue = $this->queues[$level];
            $queue->push($task);
            $this->logger->info("Task {$task->getId()} dispatched to {$level} queue");
            return true;
        } catch (Exception $e) {
            $this->logger->error("Failed to dispatch task: " . $e->getMessage());
            return $this->handleFailedDispatch($task, $level);
        }
    }
    // 确定队列级别
    private function determineQueueLevel(int $delay): string 
    {
        foreach ($this->queueLevels as $level => $config) {
            if ($delay <= $config['max_delay']) {
                return $level;
            }
        }
        return 'long'; // 默认最长队列
    }
    // 队列降级处理
    private function handleFailedDispatch(Task $task, string $failedLevel): bool 
    {
        $levels = array_keys($this->queueLevels);
        $currentIndex = array_search($failedLevel, $levels);
        // 尝试降级到更低优先级的队列
        for ($i = $currentIndex + 1; $i < count($levels); $i++) {
            $lowerLevel = $levels[$i];
            $queue = $this->queues[$lowerLevel];
            if ($queue->push($task)) {
                $this->logger->warning(
                    "Task {$task->getId()} downgraded from {$failedLevel} to {$lowerLevel}"
                );
                return true;
            }
        }
        // 所有队列都失败，持久化到数据库
        return $this->persistToDatabase($task);
    }
    // 持久化到数据库
    private function persistToDatabase(Task $task): bool 
    {
        try {
            DB::table('failed_tasks')->insert([
                'task_id' => $task->getId(),
                'task_data' => serialize($task),
                'failed_at' => date('Y-m-d H:i:s'),
                'retry_count' => 0
            ]);
            return true;
        } catch (Exception $e) {
            $this->logger->critical("Failed to persist task: " . $e->getMessage());
            return false;
        }
    }
}

消费者进程实现

<?php
class QueueConsumer 
{
    private $redis;
    private $stopFlag = false;
    private $consumerId;
    public function __construct($consumerId) 
    {
        $this->redis = new Redis();
        $this->consumerId = $consumerId;
    }
    // 启动消费者
    public function start() 
    {
        pcntl_signal(SIGTERM, function() {
            $this->stopFlag = true;
        });
        while (!$this->stopFlag) {
            $this->processQueue('high_priority');
            $this->processQueue('normal_priority');
            $this->processQueue('low_priority');
            // 检查是否有新的延时任务到期
            $this->checkDelayedTasks();
            usleep(100000); // 100ms 休息
            pcntl_signal_dispatch();
        }
    }
    // 处理队列
    private function processQueue(string $queueName) 
    {
        // 使用BLPOP阻塞获取任务
        $taskData = $this->redis->blPop(
            "queue:{$queueName}",
            0  // 无限等待
        );
        if ($taskData) {
            $task = unserialize($taskData[1]);
            $this->executeTaskWithRetry($task);
        }
    }
    // 检查延时任务
    private function checkDelayedTasks() 
    {
        $currentTime = microtime(true);
        // 从ZSET中获取到期的任务
        $expiredTasks = $this->redis->zRangeByScore(
            'delayed_queue',
            0,
            $currentTime,
            ['limit' => [0, 100]]
        );
        foreach ($expiredTasks as $taskData) {
            $task = unserialize($taskData);
            // 移出延时队列，加入立即执行队列
            $this->redis->multi();
            $this->redis->zRem('delayed_queue', $taskData);
            $this->redis->lPush('queue:immediate', serialize($task));
            $this->redis->exec();
        }
    }
    // 带重试的执行
    private function executeTaskWithRetry(Task $task, int $maxRetries = 3) 
    {
        $attempts = 0;
        while ($attempts < $maxRetries) {
            try {
                $this->executeTask($task);
                return true;
            } catch (Exception $e) {
                $attempts++;
                $this->logger->error(
                    "Task execution failed: {$task->getId()}, attempt {$attempts}"
                );
                if ($attempts < $maxRetries) {
                    $backoff = pow(2, $attempts) * 1000; // 指数退避
                    usleep($backoff * 1000);
                }
            }
        }
        // 超过最大重试次数，移到死信队列
        $this->moveToDeadLetterQueue($task);
        return false;
    }
}

监控与管理API

<?php
class QueueMonitor 
{
    // 获取队列状态
    public function getQueueStats(): array 
    {
        return [
            'total_tasks' => $this->getTotalTasks(),
            'queue_distribution' => $this->getQueueDistribution(),
            'processing_speed' => $this->getProcessingSpeed(),
            'error_rate' => $this->getErrorRate(),
            'average_latency' => $this->getAverageLatency(),
        ];
    }
    // 获取队列分布
    private function getQueueDistribution(): array 
    {
        $stats = [];
        $queues = ['immediate', 'short', 'medium', 'long'];
        foreach ($queues as $queue) {
            $stats[$queue] = [
                'pending' => $this->redis->lLen("queue:{$queue}"),
                'processing' => $this->redis->get("queue:{$queue}:processing") ?? 0,
                'delayed' => $this->redis->zCard("delayed:{$queue}"),
                'failed' => $this->redis->lLen("failed:{$queue}")
            ];
        }
        return $stats;
    }
    // 动态调整队列优先级
    public function adjustQueuePriority(string $queue, int $priority): bool 
    {
        // 更新队列配置
        $config = Config::get("queues.{$queue}");
        $config['priority'] = $priority;
        Config::set("queues.{$queue}", $config);
        // 通知消费者重新加载配置
        $this->notifyConsumers('config_reload');
        return true;
    }
}

使用示例

<?php
// 初始化队列系统
$taskManager = new DelayedTaskManager();
$dispatcher = new QueueDispatcher();
// 创建不同级别的任务
$task1 = new Task(['type' => 'email', 'data' => '...']);
$task2 = new Task(['type' => 'notification', 'data' => '...']);
$task3 = new Task(['type' => 'report', 'data' => '...']);
// 分配不同延时
$dispatcher->dispatch($task1->setDelay(5));      // 5秒：立即队列
$dispatcher->dispatch($task2->setDelay(120));    // 2分钟：短时队列
$dispatcher->dispatch($task3->setDelay(7200));   // 2小时：中时队列
// 启动消费者
$consumer = new QueueConsumer('worker_1');
$consumer->start();
// 监控
$monitor = new QueueMonitor();
$stats = $monitor->getQueueStats();
print_r($stats);

这个方案提供完整的多级延时任务队列实现,主要特点：

1. 时间轮算法：高效处理不同类型的延时需求
2. 多级存储：Redis、MySQL、文件系统混合使用
3. 自动降级：队列故障时自动降级处理
4. 可观测性：完整的监控和管理API
5. 容错机制：重试、死信队列、持久化保障

可以根据实际业务需求调整队列级别和配置参数。