PHP项目Laravel队列延迟与优先级

wen PHP项目 2026-08-12 193

本文目录导读：

延迟执行（Delayed Processing）
优先级设置（Queue Priority）
高级应用示例
监控和测试
最佳实践

在 Laravel 中，队列的延迟执行和优先级是两个重要的功能特性，让我详细介绍这两个功能的实现方法和最佳实践。

延迟执行（Delayed Processing）

基本用法

<?php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
class SendWelcomeEmail implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    public $user;
    public function __construct($user)
    {
        $this->user = $user;
    }
    public function handle()
    {
        // 执行任务
    }
}

设置延迟时间

// 方法1：在分发时指定延迟
use App\Jobs\SendWelcomeEmail;
use Illuminate\Support\Facades\Queue;
// 延迟5秒后执行
SendWelcomeEmail::dispatch($user)->delay(now()->addSeconds(5));
// 延迟10分钟
SendWelcomeEmail::dispatch($user)->delay(now()->addMinutes(10));
// 延迟到指定时间
SendWelcomeEmail::dispatch($user)->delay(now()->addDay());

在 Job 类中定义默认延迟

class SendWelcomeEmail implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    // 定义默认延迟时间
    public $delay = 60; // 60秒
    // 或者使用方法
    public function delay()
    {
        return now()->addMinutes(5);
    }
    public function handle()
    {
        // 执行任务
    }
}

条件延迟

// 根据条件设置不同延迟
$delay = $user->isNew() ? now()->addMinutes(30) : now()->addHours(1);
SendWelcomeEmail::dispatch($user)->delay($delay);

优先级设置（Queue Priority）

配置不同队列

// config/queue.php
'connections' => [
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => env('REDIS_QUEUE', 'default'),
        'retry_after' => 90,
        'block_for' => 0,
    ],
],

使用不同队列名称

// 分发到指定队列
use App\Jobs\ProcessPayment;
use App\Jobs\SendEmail;
// 高优先级队列
ProcessPayment::dispatch($order)->onQueue('high');
ProcessPayment::dispatch($order)->onQueue('high')->delay(now()->addMinutes(5));
// 普通优先级队列
SendEmail::dispatch($user)->onQueue('default');
// 低优先级队列
ProcessImage::dispatch($image)->onQueue('low');

在 Job 类中指定队列

class ProcessPayment implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    // 在构造函数中指定
    public function __construct($order)
    {
        $this->onQueue('high');
    }
    // 或使用属性
    public $queue = 'high';
    public function handle()
    {
        // 处理订单
    }
}

多队列监听配置

# 启动多个队列处理器（注意优先级顺序）
php artisan queue:work --queue=high,default,low

在 Supervisor 配置中管理优先级

; supervisor.conf
[program:laravel-high-queue]
process_name=%(program_name)s_%(process_num)02d
command=php /path/to/artisan queue:work redis --queue=high --tries=3 --max-time=3600
numprocs=2
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=forge
redirect_stderr=true
stdout_logfile=/path/to/logs/worker.log
[program:laravel-low-queue]
process_name=%(program_name)s_%(process_num)02d
command=php /path/to/artisan queue:work redis --queue=low --tries=3 --max-time=3600
numprocs=1
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=forge
redirect_stderr=true
stdout_logfile=/path/to/logs/worker.log

高级应用示例

定时任务中的延迟队列

// App\Console\Kernel.php
protected function schedule(Schedule $schedule)
{
    // 每天凌晨3点执行
    $schedule->call(function () {
        // 处理大量数据
        $users = User::where('status', 'active')->get();
        foreach ($users as $user) {
            // 分批次延迟执行
            GenerateReport::dispatch($user)
                ->onQueue('report')
                ->delay(now()->addMinutes($user->id % 60));
        }
    })->dailyAt('03:00');
}

带重试策略的延迟队列

class ProcessData implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    public $tries = 3;
    public $timeout = 120;
    public $backoff = [2, 10, 30]; // 重试间隔
    public function handle()
    {
        try {
            // 处理数据
        } catch (\Exception $e) {
            // 特定错误时重新延迟
            if ($this->attempts() < 3) {
                $this->release(now()->addMinutes(10));
            } else {
                throw $e;
            }
        }
    }
}

动态优先级调整

class ManagePriorityQueue
{
    public function adjustPriority(Job $job, $newPriority)
    {
        // 重新分配队列
        $className = get_class($job);
        $className::dispatch(...$job->getOriginal())
            ->onQueue($newPriority)
            ->delay(now()->addSeconds(5));
        // 删除原任务
        $job->delete();
    }
}

Redis 队列优先级示例

// config/queue.php - 使用 Redis 的优先级支持
'connections' => [
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => [
            'high',
            'default',
            'low',
        ],
        'retry_after' => 90,
        'block_for' => 0,
    ],
],

监控和测试

查看队列状态

# 查看队列监控
php artisan queue:monitor high,default,low
php artisan queue:failed --limit=10

在代码中监控

// 使用事件监听队列状态
Queue::before(function (JobProcessing $event) {
    Log::info('Job processing: ' . $event->job->getName());
});
Queue::after(function (JobProcessed $event) {
    Log::info('Job processed: ' . $event->job->getName());
});
Queue::failing(function (JobFailed $event) {
    Log::error('Job failed: ' . $event->job->getName());
});

测试队列

public function test_delayed_job()
{
    Queue::fake();
    SendWelcomeEmail::dispatch($user)->delay(now()->addMinutes(10));
    Queue::assertPushed(SendWelcomeEmail::class, function ($job) {
        return $job->delay->format('Y-m-d H:i') === now()->addMinutes(10)->format('Y-m-d H:i');
    });
}