Laravel多租户应用中按租户管理带优先级队列的实现方案

针对你基于Laravel 5.x、PHP 7.x和AWS搭建的多租户应用场景，我整理了一套完整的解决方案，覆盖租户独立队列、优先级插队、API限频以及AWS资源弹性释放的全部需求，下面一步步拆解实现方式：

一、租户独立队列设计：彻底隔离任务流 #

给每个租户分配专属队列名称，比如用租户ID作为后缀：tenant_{tenant_id}_queue，确保不同租户的任务不会互相干扰。

1. 任务分发时指定租户队列 #

在触发SyncCustomers、InstallCode等任务时，动态绑定对应租户的队列，同时通过后缀标识优先级：

// 低优先级：同步10000条客户数据
SyncCustomers::dispatch($tenantId)->onQueue("tenant_{$tenantId}_queue:low");

// 高优先级：install_code请求（需16次API调用）
InstallCode::dispatch($tenantId)->onQueue("tenant_{$tenantId}_queue:high");

Laravel原生支持这种带优先级后缀的队列命名，处理器会优先消费标记为high的队列任务。

二、配置优先级队列处理器：实现插队逻辑 #

我们推荐用AWS SQS作为队列驱动（天然适配AWS生态），然后配置处理器优先处理高优先级任务。

1. SQS队列基础配置 #

在config/queue.php中配置AWS SQS连接：

'connections' => [
    'sqs' => [
        'driver' => 'sqs',
        'key' => env('AWS_ACCESS_KEY_ID'),
        'secret' => env('AWS_SECRET_ACCESS_KEY'),
        'prefix' => env('SQS_PREFIX', 'https://sqs.us-east-1.amazonaws.com/your-account-id'),
        'queue' => env('SQS_QUEUE', 'default'),
        'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    ],
],

2. 启动带优先级的租户处理器 #

针对单个租户启动队列处理器时，明确指定优先级顺序（高优先级在前）：

php artisan queue:work sqs --queue=tenant_{tenant_id}_queue:high,tenant_{tenant_id}_queue:low --sleep=3 --tries=3

这样处理器会先耗尽高优先级队列的任务，再去处理低优先级的同步任务，完美实现你需要的插队效果。

三、API调用限流：严格遵守频率限制 #

为了不触发每秒2次、每分钟41次的API限制，我们可以封装一个任务基类，用Laravel缓存来控制调用频率：

1. 封装限流基类 #

创建TenantApiTask基类，所有需要调用API的任务都继承它：

abstract class TenantApiTask implements ShouldQueue
{
    protected $tenantId;
    // 对应API限制：每秒2次，每分钟41次
    protected $perSecondLimit = 2;
    protected $perMinuteLimit = 41;

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

    protected function waitForRateLimit()
    {
        $secondKey = "tenant_{$this->tenantId}_api_sec_count";
        $minuteKey = "tenant_{$this->tenantId}_api_min_count";

        // 先检查每分钟限制
        while (Cache::get($minuteKey, 0) >= $this->perMinuteLimit) {
            sleep(1);
        }

        // 再检查每秒限制
        while (Cache::get($secondKey, 0) >= $this->perSecondLimit) {
            usleep(500000); // 等待500毫秒，避免频繁轮询
        }

        // 更新计数并设置过期时间
        Cache::increment($secondKey);
        Cache::increment($minuteKey);
        Cache::put($secondKey, Cache::get($secondKey), 1);
        Cache::put($minuteKey, Cache::get($minuteKey), 60);
    }
}

2. 在任务中使用限流 #

比如InstallCode任务：

class InstallCode extends TenantApiTask
{
    public function handle()
    {
        // 循环执行16次API调用
        for ($i = 0; $i < 16; $i++) {
            $this->waitForRateLimit();
            // 执行你的API调用逻辑
            $this->callExternalApi();
        }
    }

    protected function callExternalApi()
    {
        // 这里写你的API请求代码
    }
}

四、AWS弹性扩缩容：无任务时释放资源 #

结合AWS的服务特性，我们可以实现任务负载驱动的资源弹性调整：

1. 方案一：EC2 + Auto Scaling #

用CloudWatch监控SQS队列的ApproximateNumberOfMessagesVisible指标：

• 当队列消息数为0时，自动将Auto Scaling组的实例数缩至0（或最小1个，根据你的运维需求）
• 当队列有消息时，自动扩容实例处理任务

2. 方案二：AWS Fargate + Laravel Horizon #

Laravel 5.x可以搭配Horizon 1.x版本（注意版本兼容性），配置自动扩缩容：
在config/horizon.php中添加租户队列的进程池：

'environments' => [
    'production' => [
        'tenant-pool' => [
            'connection' => 'sqs',
            'queue' => ['tenant_*_queue:high', 'tenant_*_queue:low'],
            'balance' => 'auto',
            'processes' => 10,
            'tries' => 3,
            'autoScaling' => [
                'minProcesses' => 0,
                'maxProcesses' => 20,
                'balanceCooldown' => 30,
            ],
        ],
    ],
],

Horizon会根据队列中的消息数量自动调整Fargate任务数，无任务时进程数降为0，彻底释放AWS资源。

场景验证 #

针对你提到的测试场景：

1. 租户发起10000条客户同步任务（进入低优先级队列）
2. 中途触发install_code高优先级任务
3. 队列处理器会暂停低优先级同步，先完成16次API调用，之后再继续处理同步任务
4. 全程严格遵守API频率限制，不会触发限流报错

内容的提问来源于stack exchange，提问作者Milind Patel