Laravel队列技术问询：是否可通过单个队列Worker处理所有租户数据库的任务？

完全可以实现单个队列Worker处理所有租户数据库的任务，核心思路是让Worker动态切换数据库连接，遍历所有租户的jobs表逐个处理任务。下面分享两种实用的实现方案：

方案一：自定义多租户Worker命令 #

这种方式直接替代默认的queue:work，让Worker主动轮询所有租户的数据库，适合租户数量稳定的场景。

步骤1：生成自定义命令 #

运行Artisan命令创建专属Worker：

php artisan make:command TenantQueueWorker

步骤2：编写Worker逻辑 #

在生成的app/Console/Commands/TenantQueueWorker.php中，写入以下代码：

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Illuminate\Queue\Worker;
use Illuminate\Queue\WorkerOptions;

class TenantQueueWorker extends Command
{
    protected $signature = 'tenant:queue:work {--timeout=60} {--sleep=3} {--tries=3}';
    protected $description = 'Process jobs from all tenant databases';

    public function handle()
    {
        // 从主库获取所有租户的数据库配置（这里假设你在主库有tenants表存储租户信息）
        $tenants = DB::connection('mysql')->table('tenants')->get();

        $worker = app(Worker::class);
        $options = new WorkerOptions();
        $options->timeout = $this->option('timeout');
        $options->sleep = $this->option('sleep');
        $options->tries = $this->option('tries');

        // 循环轮询所有租户，处理任务
        while (true) {
            foreach ($tenants as $tenant) {
                // 动态配置当前租户的数据库连接
                config([
                    'database.connections.tenant' => [
                        'driver' => 'mysql',
                        'host' => $tenant->db_host,
                        'database' => $tenant->db_name,
                        'username' => $tenant->db_user,
                        'password' => $tenant->db_password,
                        'charset' => 'utf8mb4',
                        'collation' => 'utf8mb4_unicode_ci',
                    ]
                ]);

                // 清除旧连接并重新连接租户数据库
                DB::purge('tenant');
                DB::connection('tenant')->reconnect();

                // 处理当前租户的下一个任务（单次处理，避免阻塞其他租户）
                $worker->runNextJob('tenant', 'default', $options);
            }

            // 所有租户轮询一遍后，短暂休眠再循环
            sleep($options->sleep);
        }
    }
}

步骤3：启动多租户Worker #

执行自定义命令启动Worker：

php artisan tenant:queue:work

方案二：租户任务分发到主队列（轻量场景） #

如果租户数量不多，也可以通过定期分发任务的方式，将所有租户的任务同步到主库队列，再用默认Worker处理：

步骤1：创建任务分发命令 #

php artisan make:command DispatchTenantJobs

步骤2：编写分发逻辑 #

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Queue;

class DispatchTenantJobs extends Command
{
    protected $signature = 'tenant:dispatch-jobs';
    protected $description = 'Dispatch jobs from all tenants to main queue';

    public function handle()
    {
        $tenants = DB::connection('mysql')->table('tenants')->get();

        foreach ($tenants as $tenant) {
            // 配置租户数据库连接
            config([
                'database.connections.tenant' => [
                    // 填写租户数据库配置
                ]
            ]);
            DB::purge('tenant');
            $tenantDb = DB::connection('tenant');

            // 获取租户未处理的任务
            $jobs = $tenantDb->table('jobs')->whereNull('reserved_at')->get();

            foreach ($jobs as $job) {
                // 将任务推送到主队列
                Queue::connection('mysql')->pushRaw($job->payload, $job->queue);
                // 删除租户库中的原任务（避免重复处理）
                $tenantDb->table('jobs')->where('id', $job->id)->delete();
            }
        }
    }
}

步骤3：调度分发任务 #

在app/Console/Kernel.php中添加定时调度，让系统自动同步任务：

protected function schedule(Schedule $schedule)
{
    // 每分钟同步一次租户任务
    $schedule->command('tenant:dispatch-jobs')->everyMinute();
}

最后启动主队列Worker即可：

php artisan queue:work

关键注意事项 #

• 连接清理：每次切换租户时必须调用DB::purge('tenant')，避免数据库连接混淆。
• 性能优化：如果租户数量较多，方案一的轮询可能有延迟，可以考虑分组处理或限制单次处理的任务数。
• 错误处理：在自定义Worker中添加异常捕获，防止单个租户的数据库错误导致整个Worker崩溃。
• 任务锁机制：确保同一任务不会被重复处理，可在任务执行前添加数据库锁。

内容的提问来源于stack exchange，提问作者Mdr Kuchhadiya