PHP队列延迟任务与定时执行

PHP队列延迟任务与定时执行延迟任务是队列系统的常见需求。订单超时取消、定时发送通知、延迟处理都可以用延迟队列实现。今天说说PHP中延迟任务的实现。用Redis的有序集合实现延迟队列。phpclass DelayedQueue{private Redis $redis;private string $prefix delay:;public function __construct(Redis $redis){$this-redis $redis;}public function push(string $queue, array $job, int $delay 0): void{$payload json_encode([id uniqid(job_, true),data $job,created_at time(),]);$executeAt time() $delay;$this-redis-zAdd($this-prefix . $queue, $executeAt, $payload);}public function later(string $queue, array $job, int $delay): void{$this-push($queue, $job, $delay);}public function pop(string $queue): ?array{$now time();$options [limit [0, 1]];$jobs $this-redis-zRangeByScore($this-prefix . $queue, 0, $now, $options);if (empty($jobs)) return null;$job $jobs[0];$this-redis-zRem($this-prefix . $queue, $job);return json_decode($job, true);}public function migrateToQueue(string $queue, string $targetQueue): int{$now time();$jobs $this-redis-zRangeByScore($this-prefix . $queue, 0, $now);$count 0;foreach ($jobs as $job) {$this-redis-lPush(queue:{$targetQueue}, $job);$this-redis-zRem($this-prefix . $queue, $job);$count;}return $count;}public function size(string $queue): int{return $this-redis-zCard($this-prefix . $queue);}public function cleanup(string $queue): int{$removed $this-redis-zRemRangeByRank($this-prefix . $queue, 0, -1001);return $removed 0 ? $removed : 0;}}$redis new Redis();$redis-connect(127.0.0.1, 6379);$delay new DelayedQueue($redis);// 添加延迟任务$delay-push(order_timeout, [order_id ORD-001, action cancel], 300);$delay-later(notifications, [to usertest.com, type reminder], 86400);echo 延迟队列大小: . $delay-size(order_timeout) . \n;// 迁移到期任务$migrated $delay-migrateToQueue(order_timeout, orders);echo 已迁移 {$migrated} 个任务\n;?基于数据库的定时任务。phpclass ScheduledTask{private PDO $pdo;public function __construct(PDO $pdo){$this-pdo $pdo;$this-initTable();}private function initTable(): void{$this-pdo-exec(CREATE TABLE IF NOT EXISTS scheduled_tasks (id INT AUTO_INCREMENT PRIMARY KEY,task_name VARCHAR(200),task_type VARCHAR(50),payload JSON,run_at TIMESTAMP,executed BOOLEAN DEFAULT FALSE,executed_at TIMESTAMP NULL,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,INDEX idx_run (run_at, executed)));}public function schedule(string $name, array $payload, DateTime $runAt): int{$stmt $this-pdo-prepare(INSERT INTO scheduled_tasks (task_name, payload, run_at)VALUES (?, ?, ?));$stmt-execute([$name, json_encode($payload), $runAt-format(Y-m-d H:i:s)]);return (int)$this-pdo-lastInsertId();}public function processDue(): int{$stmt $this-pdo-prepare(SELECT * FROM scheduled_tasksWHERE run_at NOW() AND executed FALSELIMIT 50);$stmt-execute();$tasks $stmt-fetchAll();$count 0;foreach ($tasks as $task) {$this-pdo-prepare(UPDATE scheduled_tasks SET executed TRUE, executed_at NOW() WHERE id ?)-execute([$task[id]]);$count;}return $count;}}?延时任务在业务中很常用。Redis的zset实现简单高效数据库实现可靠但没有自动过期机制。选择合适的实现方式取决于对可靠性的要求。