任务队列 #

队列处理流程
┌─────────────┐
│   用户请求   │
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  分发任务    │
│  到队列      │
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  立即返回    │
│  响应        │
└─────────────┘

       ↓ 异步

┌─────────────┐
│  队列Worker  │
│  处理任务    │
└─────────────┘

// config/queue.php
'default' => env('QUEUE_CONNECTION', 'database'),

'connections' => [
    'sync' => [
        'driver' => 'sync',
    ],
    
    'database' => [
        'driver' => 'database',
        'table' => 'jobs',
        'queue' => 'default',
        'retry_after' => 90,
        'after_commit' => true,
    ],
    
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => env('REDIS_QUEUE', 'default'),
        'retry_after' => 90,
        'block_for' => null,
        'after_commit' => true,
    ],
],

php artisan queue:table
php artisan migrate

php artisan make:job SendEmail

// app/Jobs/SendEmail.php
namespace App\Jobs;

use App\Models\User;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Mail;

class SendEmail implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $user;

    public function __construct(User $user)
    {
        $this->user = $user;
    }

    public function handle()
    {
        Mail::to($this->user->email)->send(new WelcomeEmail($this->user));
    }
}

class SendEmail implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public $tries = 3;           // 最大重试次数
    public $maxExceptions = 3;   // 最大异常次数
    public $timeout = 120;       // 超时时间（秒）
    public $backoff = [10, 30, 60]; // 重试间隔
    
    // 或使用方法
    public function retryUntil()
    {
        return now()->addMinutes(30);
    }
    
    public function middleware()
    {
        return [
            new WithoutOverlapping,
            new ThrottleRequests(10),
        ];
    }
}

use App\Jobs\SendEmail;

SendEmail::dispatch($user);

// 延迟指定时间
SendEmail::dispatch($user)->delay(now()->addMinutes(10));

// 延迟指定秒数
SendEmail::dispatch($user)->delay(60);

SendEmail::dispatch($user)->onQueue('emails');

// 在任务类中指定
class SendEmail implements ShouldQueue
{
    public $queue = 'emails';
}

SendEmail::dispatch($user)->onConnection('redis');

if ($this->user->subscribed()) {
    SendEmail::dispatch($this->user);
}

SendEmail::dispatchSync($user);

php artisan queue:work

# 指定连接和队列
php artisan queue:work redis --queue=high,default

# 守护进程模式
php artisan queue:work --daemon

# 指定超时和内存
php artisan queue:work --timeout=60 --memory=128

# 指定尝试次数
php artisan queue:work --tries=3

# 延迟重试
php artisan queue:work --backoff=10

# 处理指定数量任务后停止
php artisan queue:work --max-jobs=1000

# 处理指定时间后停止
php artisan queue:work --max-time=3600

php artisan queue:listen

php artisan queue:work --once

php artisan queue:failed-table
php artisan migrate

class SendEmail implements ShouldQueue
{
    public function failed(\Throwable $exception)
    {
        // 记录日志
        Log::error('邮件发送失败', [
            'user' => $this->user->id,
            'error' => $exception->getMessage(),
        ]);
        
        // 通知管理员
        Mail::to('admin@example.com')->send(new JobFailedMail($this->user));
    }
}

# 重试所有失败任务
php artisan queue:retry all

# 重试指定任务
php artisan queue:retry 5

# 重试多个任务
php artisan queue:retry 5 6 7

# 删除指定任务
php artisan queue:forget 5

# 删除所有失败任务
php artisan queue:flush

// app/Jobs/Middleware/RateLimited.php
namespace App\Jobs\Middleware;

use Illuminate\Support\Facades\Redis;

class RateLimited
{
    public function handle($job, $next)
    {
        Redis::throttle('key')
            ->allow(10)->every(60)
            ->then(function () use ($job, $next) {
                $next($job);
            }, function () use ($job) {
                $job->release(10);
            });
    }
}

class SendEmail implements ShouldQueue
{
    public function middleware()
    {
        return [
            new RateLimited,
        ];
    }
}

use Illuminate\Queue\Middleware\WithoutOverlapping;
use Illuminate\Queue\Middleware\ThrottleRequests;
use Illuminate\Queue\Middleware\SkipIfBatchCancelled;

public function middleware()
{
    return [
        (new WithoutOverlapping)->releaseAfter(60),
        new ThrottleRequests(10),
        new SkipIfBatchCancelled,
    ];
}

php artisan queue:batches-table
php artisan migrate

use Illuminate\Bus\Batchable;

class ImportCsv implements ShouldQueue
{
    use Batchable;

    public function handle()
    {
        if ($this->batch()->cancelled()) {
            return;
        }

        // 处理数据
    }
}

use Illuminate\Support\Facades\Bus;
use App\Jobs\ImportCsv;

$batch = Bus::batch([
    new ImportCsv(1, 100),
    new ImportCsv(101, 200),
    new ImportCsv(201, 300),
])->then(function (Batch $batch) {
    // 所有任务完成
})->catch(function (Batch $batch, Throwable $e) {
    // 任务失败
})->finally(function (Batch $batch) {
    // 无论成功失败都执行
})->dispatch();

composer require laravel/horizon
php artisan horizon:install

// config/horizon.php
'environments' => [
    'production' => [
        'supervisor-1' => [
            'maxProcesses' => 10,
            'balanceMaxShift' => 1,
            'balanceCooldown' => 3,
        ],
    ],
    
    'local' => [
        'supervisor-1' => [
            'maxProcesses' => 3,
        ],
    ],
],

php artisan horizon