Laravel 13 队列路由详解

摘要

Laravel 13 引入了 Queue::route() 方法，允许在中央位置定义特定任务的默认队列/连接路由规则。本文将深入讲解队列路由的使用，包括：

  • 队列路由核心概念
  • Queue::route() 基础用法
  • 任务分类与路由策略
  • 动态路由配置
  • 与队列属性的配合
  • 实战案例：多队列系统

本文适合希望优化队列管理的 Laravel 开发者。

1. 队列路由概述

1.1 传统方式的问题

在 Laravel 13 之前，任务队列配置分散在多个地方：

1
2
3
4
5
6
7
8
9
// 在任务类中
class ProcessPodcast implements ShouldQueue
{
    public $connection = 'redis';
    public $queue = 'podcasts';
}

// 或在调度时指定
ProcessPodcast::dispatch()->onConnection('redis')->onQueue('podcasts');

这种方式的问题：

  • 配置分散，难以统一管理
  • 修改需要改动多处代码
  • 难以实现动态路由

1.2 Queue::route() 的优势

1
2
// 在 AppServiceProvider 中统一配置
Queue::route(ProcessPodcast::class, connection: 'redis', queue: 'podcasts');

优势：

  • 配置集中管理
  • 易于修改和维护
  • 支持动态路由
  • 与属性配合使用

2. 基础用法

2.1 基本路由

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<?php

namespace App\Providers;

use Illuminate\Support\ServiceProvider;
use Illuminate\Support\Facades\Queue;
use App\Jobs\ProcessPodcast;
use App\Jobs\SendEmail;
use App\Jobs\GenerateReport;

class AppServiceProvider extends ServiceProvider
{
    public function boot(): void
    {
        Queue::route(ProcessPodcast::class, connection: 'redis', queue: 'podcasts');
        Queue::route(SendEmail::class, connection: 'database', queue: 'emails');
        Queue::route(GenerateReport::class, connection: 'redis', queue: 'reports');
    }
}

2.2 只指定队列

1
Queue::route(ProcessPodcast::class, queue: 'podcasts');

2.3 只指定连接

1
Queue::route(ProcessPodcast::class, connection: 'redis');

2.4 批量路由

1
2
3
4
5
6
7
8
9
use App\Jobs\ProcessPodcast;
use App\Jobs\ProcessVideo;
use App\Jobs\ProcessAudio;

Queue::route([
    ProcessPodcast::class,
    ProcessVideo::class,
    ProcessAudio::class,
], connection: 'redis', queue: 'media');

3. 路由策略

3.1 按优先级路由

1
2
3
4
5
6
7
8
9
10
11
// 高优先级队列
Queue::route(ProcessPayment::class, queue: 'high');
Queue::route(SendNotification::class, queue: 'high');

// 普通优先级队列
Queue::route(GenerateReport::class, queue: 'default');
Queue::route(ProcessImage::class, queue: 'default');

// 低优先级队列
Queue::route(CleanupData::class, queue: 'low');
Queue::route(ArchiveLogs::class, queue: 'low');

3.2 按资源类型路由

1
2
3
4
5
6
7
8
9
// CPU 密集型任务
Queue::route(ProcessVideo::class, queue: 'cpu-intensive');
Queue::route(ProcessImage::class, queue: 'cpu-intensive');
Queue::route(GeneratePdf::class, queue: 'cpu-intensive');

// I/O 密集型任务
Queue::route(SendEmail::class, queue: 'io-intensive');
Queue::route(SyncExternalApi::class, queue: 'io-intensive');
Queue::route(UploadFile::class, queue: 'io-intensive');

3.3 按连接类型路由

1
2
3
4
5
6
7
8
// Redis 连接（高性能）
Queue::route(ProcessPodcast::class, connection: 'redis', queue: 'podcasts');

// Database 连接（持久化）
Queue::route(ProcessOrder::class, connection: 'database', queue: 'orders');

// SQS 连接（云服务）
Queue::route(ProcessExternalWebhook::class, connection: 'sqs', queue: 'webhooks');

4. 动态路由

4.1 基于条件路由

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
use App\Jobs\ProcessPodcast;
use Illuminate\Support\Facades\Queue;

class AppServiceProvider extends ServiceProvider
{
    public function boot(): void
    {
        Queue::route(ProcessPodcast::class, function ($job) {
            if ($job->isPriority()) {
                return ['connection' => 'redis', 'queue' => 'high'];
            }
            
            return ['connection' => 'redis', 'queue' => 'default'];
        });
    }
}

4.2 基于环境路由

1
2
3
4
5
6
7
Queue::route(ProcessPodcast::class, function ($job) {
    if (app()->environment('production')) {
        return ['connection' => 'sqs', 'queue' => 'production-podcasts'];
    }
    
    return ['connection' => 'database', 'queue' => 'local-podcasts'];
});

4.3 基于负载路由

1
2
3
4
5
6
7
8
9
10
11
use Illuminate\Support\Facades\Cache;

Queue::route(ProcessVideo::class, function ($job) {
    $load = Cache::get('queue:load:cpu-intensive', 0);
    
    if ($load > 80) {
        return ['connection' => 'redis', 'queue' => 'cpu-intensive-overflow'];
    }
    
    return ['connection' => 'redis', 'queue' => 'cpu-intensive'];
});

5. 与队列属性配合

5.1 属性优先级

1
2
3
4
5
6
7
8
9
10
11
12
13
use Illuminate\Queue\Attributes\OnQueue;
use Illuminate\Queue\Attributes\OnConnection;

#[OnConnection('database')]
#[OnQueue('emails')]
class SendEmail implements ShouldQueue
{
    // 属性定义会被路由覆盖
}

// 路由配置
Queue::route(SendEmail::class, connection: 'redis', queue: 'priority-emails');
// 最终使用: redis 连接, priority-emails 队列

5.2 属性作为默认值

1
2
3
4
5
6
7
8
#[OnQueue('default')]
class ProcessPodcast implements ShouldQueue
{
    // 如果没有路由配置，使用 'default' 队列
}

// 只有特定场景覆盖
Queue::route(ProcessPodcast::class, queue: 'priority-podcasts');

6. 实战案例

6.1 多租户队列系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
<?php

namespace App\Providers;

use Illuminate\Support\ServiceProvider;
use Illuminate\Support\Facades\Queue;
use App\Jobs\ProcessTenantTask;

class QueueServiceProvider extends ServiceProvider
{
    public function boot(): void
    {
        Queue::route(ProcessTenantTask::class, function ($job) {
            $tenant = $job->getTenant();
            
            return [
                'connection' => 'redis',
                'queue' => "tenant-{$tenant->id}",
            ];
        });
    }
}

6.2 分级队列系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
<?php

namespace App\Services;

use Illuminate\Support\Facades\Queue;
use App\Jobs\ProcessOrder;

class OrderQueueService
{
    public function configureRouting(): void
    {
        Queue::route(ProcessOrder::class, function ($job) {
            $order = $job->getOrder();
            
            // VIP 客户订单优先处理
            if ($order->customer->isVip()) {
                return ['queue' => 'vip-orders'];
            }
            
            // 大额订单单独队列
            if ($order->total > 10000) {
                return ['queue' => 'large-orders'];
            }
            
            // 普通订单
            return ['queue' => 'orders'];
        });
    }
}

6.3 监控与告警

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
<?php

namespace App\Providers;

use Illuminate\Support\ServiceProvider;
use Illuminate\Support\Facades\Queue;
use Illuminate\Support\Facades\Log;

class QueueMonitorServiceProvider extends ServiceProvider
{
    public function boot(): void
    {
        Queue::routing(function ($job, $route) {
            Log::info('Job routed', [
                'job' => get_class($job),
                'connection' => $route['connection'] ?? 'default',
                'queue' => $route['queue'] ?? 'default',
            ]);
        });
    }
}

7. 配置文件

7.1 队列路由配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// config/queue.php
return [
    'routing' => [
        'enabled' => true,
        
        'rules' => [
            \App\Jobs\ProcessPodcast::class => [
                'connection' => 'redis',
                'queue' => 'podcasts',
            ],
            \App\Jobs\SendEmail::class => [
                'connection' => 'database',
                'queue' => 'emails',
            ],
        ],
    ],
];

7.2 加载配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
<?php

namespace App\Providers;

use Illuminate\Support\ServiceProvider;
use Illuminate\Support\Facades\Queue;

class QueueServiceProvider extends ServiceProvider
{
    public function boot(): void
    {
        if (!config('queue.routing.enabled')) {
            return;
        }
        
        foreach (config('queue.routing.rules') as $job => $route) {
            Queue::route($job, $route);
        }
    }
}

8. 测试

8.1 测试路由配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
<?php

namespace Tests\Feature;

use Tests\TestCase;
use Illuminate\Support\Facades\Queue;
use App\Jobs\ProcessPodcast;

class QueueRoutingTest extends TestCase
{
    public function test_job_is_routed_correctly()
    {
        Queue::fake();
        
        ProcessPodcast::dispatch($podcast);
        
        Queue::assertPushedOn('podcasts', ProcessPodcast::class);
        Queue::assertPushedWithConnection('redis', ProcessPodcast::class);
    }
}

8.2 测试动态路由

1
2
3
4
5
6
7
8
9
10
public function test_dynamic_routing()
{
    Queue::fake();
    
    $vipOrder = Order::factory()->vip()->create();
    
    ProcessOrder::dispatch($vipOrder);
    
    Queue::assertPushedOn('vip-orders', ProcessOrder::class);
}

9. 最佳实践

9.1 集中配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 推荐：在 ServiceProvider 中集中配置
class QueueServiceProvider extends ServiceProvider
{
    public function boot(): void
    {
        $this->configureMediaQueue();
        $this->configureEmailQueue();
        $this->configureReportQueue();
    }
    
    private function configureMediaQueue(): void
    {
        Queue::route([
            ProcessVideo::class,
            ProcessAudio::class,
            ProcessImage::class,
        ], connection: 'redis', queue: 'media');
    }
}

9.2 命名规范

1
2
3
4
5
6
// 推荐：使用描述性队列名
Queue::route(ProcessPodcast::class, queue: 'media-podcasts');
Queue::route(ProcessVideo::class, queue: 'media-videos');

// 不推荐
Queue::route(ProcessPodcast::class, queue: 'q1');

9.3 监控与日志

1
2
3
4
5
6
7
8
Queue::routing(function ($job, $route) {
    if (app()->environment('production')) {
        Log::channel('queue')->info('Job routed', [
            'job' => get_class($job),
            'route' => $route,
        ]);
    }
});

10. 总结

Laravel 13 的队列路由功能为队列管理提供了更灵活的方式：

  1. 集中配置：所有路由规则统一管理
  2. 动态路由：支持基于条件的路由决策
  3. 易于维护：修改路由无需改动任务类
  4. 与属性配合：灵活的优先级控制

通过本指南，您已经掌握了队列路由的核心用法，可以开始优化应用的队列管理了。

参考资料