队列与任务调度

队列（Queue）和任务调度（Task Scheduling）是现代 PHP 应用处理耗时操作和定时任务的两大核心机制。队列将耗时任务（如发送邮件、处理图片、推送通知）异步化，避免阻塞 HTTP 请求；任务调度则允许在指定时间自动执行定期任务（如数据清理、报表生成、缓存刷新）。本节将全面介绍队列和任务调度的概念、原理和各种实现方案。

前置知识

阅读本节前，建议先了解：

PHP 进程控制（参见 pcntl 扩展）
事件系统（参见事件系统）
Composer 基本命令（参见 Composer 基本命令）

基础概念

为什么要使用队列

没有队列：
  用户请求 → 发送邮件（3秒）→ 处理图片（5秒）→ 推送通知（2秒）→ 返回响应（10秒）

有队列：
  用户请求 → 推入队列 → 立即返回响应（50ms）
                 ↓
            后台 Worker 处理 → 发送邮件 → 处理图片 → 推送通知

队列架构

生产者（Producer）
    → 将任务推入队列

队列（Queue/Broker）
    → 存储等待处理的任务
    → 支持多种后端：Redis、RabbitMQ、数据库

消费者（Consumer/Worker）
    → 从队列拉取任务
    → 执行任务
    → 重试失败的任务

详细说明

1. 任务定义

php

<?php
declare(strict_types=1);

namespace App\Jobs;

use App\Services\EmailService;
use Psr\Log\LoggerInterface;

class SendWelcomeEmail
{
    public function __construct(
        public readonly int $userId,
        public readonly string $email,
        public readonly string $name,
    ) {}

    /**
     * 执行任务
     */
    public function handle(EmailService $emailService, LoggerInterface $logger): void
    {
        try {
            $emailService->send(
                to: $this->email,
                subject: "欢迎加入，{$this->name}！",
                template: 'emails.welcome',
                data: ['name' => $this->name],
            );

            $logger->info('Welcome email sent', [
                'user_id' => $this->userId,
            ]);
        } catch (\Throwable $e) {
            $logger->error('Failed to send welcome email', [
                'user_id' => $this->userId,
                'error' => $e->getMessage(),
            ]);

            throw $e; // 重新抛出以触发队列重试
        }
    }

    /**
     * 任务失败时的回调
     */
    public function failed(\Throwable $e, LoggerInterface $logger): void
    {
        $logger->error('Welcome email job failed permanently', [
            'user_id' => $this->userId,
            'error' => $e->getMessage(),
        ]);
    }
}

2. 队列接口设计

php

<?php
declare(strict_types=1);

namespace App\Queue;

interface QueueInterface
{
    /**
     * 推送任务到队列
     */
    public function push(string $queue, string $jobClass, array $payload = [], array $options = []): string;

    /**
     * 延迟推送任务
     */
    public function later(int $delay, string $queue, string $jobClass, array $payload = []): string;

    /**
     * 拉取任务
     */
    public function pop(string $queue): ?QueueJob;

    /**
     * 确认任务完成
     */
    public function ack(string $jobId): void;

    /**
     * 标记任务失败
     */
    public function fail(string $jobId, string $error): void;
}

interface QueueJob
{
    public function getId(): string;
    public function getJobClass(): string;
    public function getPayload(): array;
    public function attempts(): int;
    public function markAsAttempted(): void;
}

3. Redis 队列实现

php

<?php
declare(strict_types=1);

namespace App\Queue;

use Redis;

class RedisQueue implements QueueInterface
{
    public function __construct(
        private readonly Redis $redis,
        private readonly string $prefix = 'queue:',
        private readonly int $retryAfter = 30,
        private readonly int $maxTries = 3,
    ) {}

    public function push(
        string $queue,
        string $jobClass,
        array $payload = [],
        array $options = []
    ): string {
        $jobId = uniqid('job_', true);
        $job = json_encode([
            'id' => $jobId,
            'class' => $jobClass,
            'payload' => $payload,
            'attempts' => 0,
            'created_at' => time(),
        ]);

        $this->redis->rpush($this->prefix . $queue, $job);

        return $jobId;
    }

    public function later(
        int $delay,
        string $queue,
        string $jobClass,
        array $payload = []
    ): string {
        $jobId = $this->push($queue, $jobClass, $payload);
        $this->redis->zadd($this->prefix . $queue . ':delayed', time() + $delay, $jobId);
        return $jobId;
    }

    public function pop(string $queue): ?QueueJob
    {
        $raw = $this->redis->lpop($this->prefix . $queue);
        if ($raw === false) {
            return null;
        }

        $data = json_decode($raw, true);
        return new RedisQueueJob($data);
    }

    public function ack(string $jobId): void
    {
        $this->redis->del($this->prefix . 'reserved:' . $jobId);
    }

    public function fail(string $jobId, string $error): void
    {
        // 实现失败逻辑
    }
}

4. 队列 Worker

php

<?php
declare(strict_types=1);

namespace App\Queue;

use Psr\Log\LoggerInterface;

class QueueWorker
{
    private bool $shouldStop = false;

    public function __construct(
        private readonly QueueInterface $queue,
        private readonly ContainerInterface $container,
        private readonly LoggerInterface $logger,
        private readonly int $sleepSeconds = 3,
        private readonly int $maxTries = 3,
    ) {}

    /**
     * 处理队列任务
     */
    public function run(string $queue, int $maxJobs = 1000): void
    {
        $this->registerSignalHandlers();

        $jobsProcessed = 0;

        while (!$this->shouldStop && $jobsProcessed < $maxJobs) {
            $job = $this->queue->pop($queue);

            if ($job === null) {
                sleep($this->sleepSeconds);
                continue;
            }

            $this->processJob($job);
            $jobsProcessed++;
        }

        $this->logger->info('Worker stopped', [
            'jobs_processed' => $jobsProcessed,
        ]);
    }

    private function processJob(QueueJob $job): void
    {
        try {
            $jobClass = $job->getJobClass();
            $payload = $job->getPayload();

            $instance = $this->container->get($jobClass);
            $instance->handle(...array_values($payload));

            $this->queue->ack($job->getId());

            $this->logger->info('Job processed', ['job_id' => $job->getId()]);
        } catch (\Throwable $e) {
            $job->markAsAttempted();

            if ($job->attempts() >= $this->maxTries) {
                $this->queue->fail($job->getId(), $e->getMessage());
                $this->logger->error('Job failed permanently', [
                    'job_id' => $job->getId(),
                    'error' => $e->getMessage(),
                ]);
            } else {
                $this->logger->warning('Job failed, will retry', [
                    'job_id' => $job->getId(),
                    'attempts' => $job->attempts(),
                    'error' => $e->getMessage(),
                ]);
                // 重新推入队列
                sleep(5);
                $this->queue->push('default', $job->getJobClass(), $job->getPayload());
            }
        }
    }

    private function registerSignalHandlers(): void
    {
        pcntl_async_signals(true);
        pcntl_signal(SIGTERM, fn() => $this->shouldStop = true);
        pcntl_signal(SIGINT, fn() => $this->shouldStop = true);
    }
}

5. 任务调度

php

<?php
declare(strict_types=1);

namespace App\Scheduler;

interface ScheduleInterface
{
    public function addJob(
        string $jobClass,
        string $cronExpression,
        string $queue = 'default'
    ): void;
}

class CronScheduler implements ScheduleInterface
{
    /** @var array<string, array{job: string, cron: string, queue: string}> */
    private array $scheduled = [];

    public function addJob(string $jobClass, string $cronExpression, string $queue = 'default'): void
    {
        $this->scheduled[$jobClass] = [
            'job' => $jobClass,
            'cron' => $cronExpression,
            'queue' => $queue,
        ];
    }

    /**
     * 运行到期的任务
     */
    public function runDueJobs(QueueInterface $queue, LoggerInterface $logger): void
    {
        $now = time();

        foreach ($this->scheduled as $name => $config) {
            if ($this->isDue($config['cron'], $now)) {
                $logger->info('Running scheduled job', ['job' => $name]);
                $queue->push($config['queue'], $config['job']);
            }
        }
    }

    private function isDue(string $cronExpression, int $timestamp): bool
    {
        // 简化版：实际应使用 cron 表达式解析库
        // 如 dragonmantank/cron-expression
        return true;
    }
}

php

<?php
// 使用 Cron 表达式定义调度任务
$scheduler = new CronScheduler();

$scheduler->addJob(CleanTempFiles::class, '0 3 * * *');        // 每天凌晨 3 点
$scheduler->addJob(GenerateDailyReport::class, '0 8 * * 1-5');  // 工作日每天 8 点
$scheduler->addJob(RefreshCache::class, '*/15 * * * *');        // 每 15 分钟
$scheduler->addJob(DatabaseBackup::class, '0 2 * * 0');       // 每周日凌晨 2 点
$scheduler->addJob(SendWeeklyNewsletter::class, '0 9 * * 1');    // 每周一 9 点

6. Cron 表达式语法

┌────────── 分钟（0-59）
│ ┌──────── 小时（0-23）
│ │ ┌────── 日（1-31）
│ │ │ ┌──── 月份（1-12）
│ │ │ │ ┌── 星期（0-6，0=周日）
│ │ │ │ │
* * * * *

常见表达式：
  */5 * * * *    每 5 分钟
  0 * * * *      每小时整点
  0 0 * * *      每天午夜
  0 8 * * 1-5    工作日每天 8 点
  0 2 * * 0      每周日凌晨 2 点
  0 0 1 * *      每月 1 号午夜
  0 0 1 1 *      每年 1 月 1 日午夜

实战示例

场景一：多队列 Worker 部署

bash

# 运行默认队列的 Worker
php worker.php --queue=default

# 运行高优先级队列的 Worker
php worker.php --queue=high

# 运行邮件队列的 Worker（低优先级）
php worker.php --queue=emails

# 使用 Supervisor 管理进程
# /etc/supervisor/conf.d/worker.conf
# [program:laravel-worker]
# process_name=%(program_name)s_%(process_num)02d
# command=php /path/to/worker.php --queue=default
# autostart=true
# autorestart=true
# user=www-data
# numprocs=2
# redirect_stderr=true
# stdout_logfile=/path/to/worker.log

场景二：延迟队列与定时任务

php

<?php
declare(strict_types=1);

// 延迟推送任务
$queue->later(3600, 'default', SendReminderEmail::class, [
    'userId' => 42,
    'email' => 'user@example.com',
]);

// 5 分钟后执行
$queue->later(300, 'default', CheckPaymentStatus::class, [
    'orderId' => 1001,
]);

// 组合任务链
class OrderProcessingChain
{
    public function __construct(
        private readonly QueueInterface $queue,
    ) {}

    public function process(int $orderId): void
    {
        $this->queue->later(0, 'default', ValidateOrder::class, ['orderId' => $orderId]);
        $this->queue->later(60, 'default', ProcessPayment::class, ['orderId' => $orderId]);
        $this->queue->later(120, 'default', SendConfirmation::class, ['orderId' => $orderId]);
        $this->queue->later(180, 'default', UpdateInventory::class, ['orderId' => $orderId]);
    }
}

场景三：失败任务处理

php

<?php
declare(strict_types=1);

namespace App\Jobs;

class ProcessImport implements \ArrayAccess
{
    public int $tries = 3;
    public int $backoff = 30;        // 重试间隔（秒）
    public int $timeout = 120;       // 超时时间（秒）
    public int $maxExceptions = 1;    // 最大异常数

    public function __construct(
        public readonly string $filePath,
        public readonly int $userId,
    ) {}

    public function handle(): void
    {
        // 处理导入逻辑
    }

    /**
     * 计算重试间隔（指数退避）
     */
    public function retryAfter(): int
    {
        return 30 * pow(2, $this->attempts - 1);
    }

    /**
     * 任务失败回调
     */
    public function failed(\Throwable $exception): void
    {
        // 通知管理员
        // 记录错误日志
        // 标记导入为失败状态
    }
}

场景四：任务监控

php

<?php
declare(strict_types=1);

namespace App\Console\Commands;

use Illuminate\Console\Command;

class MonitorQueueCommand extends Command
{
    protected $signature = 'queue:monitor {queue=default} {--max=100}';

    protected $description = 'Monitor queue health';

    public function handle(): void
    {
        $queue = $this->argument('queue');
        $max = (int) $this->option('max');
        $size = $this->getQueueSize($queue);

        $this->table(
            ['Queue', 'Size', 'Status', 'Time'],
            [[
                $queue,
                $size,
                $size > $max ? 'CRITICAL' : 'OK',
                now()->toDateTimeString(),
            ]]
        );

        if ($size > $max) {
            $this->error("Queue {$queue} has {$size} jobs (max: {$max})");
            // 发送告警
        }
    }

    private function getQueueSize(string $queue): int
    {
        // 根据队列驱动获取队列大小
        return 0;
    }
}

注意事项

1. 任务幂等性

php

<?php
declare(strict_types=1);

class ProcessPayment
{
    public function handle(int $orderId): void
    {
        // ✅ 幂等：检查订单状态
        $order = Order::findOrFail($orderId);
        if ($order->isPaid()) {
            return; // 已处理过，直接返回
        }

        $order->markAsPaid();
        // ...
    }
}

2. 任务监控

监控队列长度（积压任务数）
监控 Worker 进程状态
监控任务失败率
设置告警阈值

最佳实践

1. 任务大小控制

好的任务：
- 发送邮件（几秒）
- 生成缩略图（几百毫秒）
- 推送通知（几百毫秒）

不好的任务：
- 导出百万行数据（应拆分为小任务）
- 视频转码（几分钟，应使用专用服务）

继续学习：ORM 基础 — 了解对象关系映射的基本概念。

队列与任务调度 ​

基础概念 ​

为什么要使用队列 ​

队列架构 ​

详细说明 ​

1. 任务定义 ​

2. 队列接口设计 ​

3. Redis 队列实现 ​

4. 队列 Worker ​

5. 任务调度 ​

6. Cron 表达式语法 ​

实战示例 ​

场景一：多队列 Worker 部署 ​

场景二：延迟队列与定时任务 ​

场景三：失败任务处理 ​

场景四：任务监控 ​

注意事项 ​

1. 任务幂等性 ​

2. 任务监控 ​

最佳实践 ​

1. 任务大小控制 ​

参考链接 ​