多租户Yii2应用：按租户定时并行执行Cron任务的实现问题

针对你遇到的多租户定时任务并行执行问题，核心思路是放弃给每个租户单独配置触发器，改用「统一调度器+分布式队列」的模式，既支持租户动态配置定时规则，又能实现任务并行执行。以下是两种适配Yii2技术栈的可行方案：

方案1：AWS EventBridge + SQS + Yii2 Worker（推荐） #

解决痛点： #

• 用单条EventBridge规则替代静态多规则，解决动态租户配置的问题
• 通过SQS多消费者实现任务并行，弥补你之前用SQS时未启用多worker的不足

实现步骤： #

1. 配置EventBridge主触发器：创建一条每分钟触发一次的规则，目标指向Yii2控制台命令（或Lambda函数，推荐直接用Yii2命令简化逻辑）。
2. 定时规则扫描：控制台命令查询所有租户的定时配置（存储在MySQL的tenant_config表，字段含id, cron_expression, last_executed），用Cron表达式判断当前时间是否需要执行该租户的同步任务。
3. 任务分发到SQS：对符合条件的租户，将租户ID、同步参数等封装成消息发送到SQS队列。
4. 多Worker并行消费：用Supervisor启动多个Yii2 Worker进程，持续从SQS拉取消息并执行同步逻辑。

Yii2调度器命令示例：

<?php
namespace app\commands;

use yii\console\Controller;
use app\models\TenantConfig;
use Cron\CronExpression;
use Aws\Sqs\SqsClient;

class TenantSchedulerController extends Controller
{
    public function actionRun()
    {
        $sqs = new SqsClient([
            'region' => 'your-region',
            'version' => 'latest',
            'credentials' => [
                'key' => 'your-aws-key',
                'secret' => 'your-aws-secret',
            ],
        ]);
        $queueUrl = 'your-sqs-queue-url';
        
        $tenants = TenantConfig::find()->all();
        $now = new \DateTime();
        
        foreach ($tenants as $tenant) {
            $cron = CronExpression::factory($tenant->cron_expression);
            if ($cron->isDue($now)) {
                // 避免1分钟内重复触发
                if ($tenant->last_executed && $tenant->last_executed->diff($now)->i < 1) {
                    continue;
                }
                
                $sqs->sendMessage([
                    'QueueUrl' => $queueUrl,
                    'MessageBody' => json_encode([
                        'tenant_id' => $tenant->id,
                        'sync_time' => $now->format('Y-m-d H:i:s')
                    ])
                ]);
                
                $tenant->last_executed = $now;
                $tenant->save(false);
            }
        }
        
        $this->stdout('Scheduler run completed.' . PHP_EOL);
    }
}

Yii2 Worker命令示例：

<?php
namespace app\commands;

use yii\console\Controller;
use Aws\Sqs\SqsClient;
use app\services\ThirdPartySyncService;

class TenantWorkerController extends Controller
{
    public function actionRun()
    {
        $sqs = new SqsClient([
            'region' => 'your-region',
            'version' => 'latest',
            'credentials' => [
                'key' => 'your-aws-key',
                'secret' => 'your-aws-secret',
            ],
        ]);
        $queueUrl = 'your-sqs-queue-url';
        $syncService = new ThirdPartySyncService();
        
        $this->stdout('Worker started. Listening for messages...' . PHP_EOL);
        
        while (true) {
            $result = $sqs->receiveMessage([
                'QueueUrl' => $queueUrl,
                'WaitTimeSeconds' => 20, // 长轮询减少空请求
                'MaxNumberOfMessages' => 1
            ]);
            
            if (!empty($result['Messages'])) {
                foreach ($result['Messages'] as $message) {
                    $body = json_decode($message['Body'], true);
                    $tenantId = $body['tenant_id'];
                    
                    try {
                        // 执行第三方API同步逻辑
                        $syncService->syncEmployeeAndPosition($tenantId);
                        
                        // 删除已处理消息
                        $sqs->deleteMessage([
                            'QueueUrl' => $queueUrl,
                            'ReceiptHandle' => $message['ReceiptHandle']
                        ]);
                        
                        $this->stdout("Tenant {$tenantId} sync completed." . PHP_EOL);
                    } catch (\Exception $e) {
                        $this->stderr("Tenant {$tenantId} sync failed: " . $e->getMessage() . PHP_EOL);
                        // 可根据错误类型决定是否重新入队
                    }
                }
            }
        }
    }
}

并行配置：

用Supervisor配置多个Worker进程，示例配置：

[program:yii-tenant-worker]
command=/path/to/php /path/to/yii tenant-worker/run
autostart=true
autorestart=true
numprocs=5 ; 启动5个并行进程
user=www-data
redirect_stderr=true
stdout_logfile=/var/log/yii-worker.log

方案2：Yii2 + Redis队列 + Supervisor（无AWS依赖） #

如果不想依赖AWS服务，可改用Redis作为队列：

1. 同样用每分钟执行的Yii2调度器命令，扫描租户定时配置，将任务推送到Redis列表。
2. 启动多个Supervisor管理的Worker进程，从Redis列表弹出任务并执行同步逻辑。

核心优势：租户配置更新后无需修改任何触发器，调度器会自动识别新的定时规则。

关键注意事项 #

• 租户数据隔离：所有同步到MySQL的业务表必须添加tenant_id字段，确保数据不会跨租户混淆。
• 幂等性保障：同步任务要支持重复执行，比如用「员工ID+租户ID」作为唯一键，或每次同步前清理该租户的旧数据。
• 错误重试：SQS自带消息可见性超时机制，失败消息会自动重回队列；Redis队列可自行实现重试次数限制，超过阈值的任务存入死信队列。
• 监控告警：对Worker进程添加监控，比如采集执行成功率、耗时数据，配置告警机制及时发现异常。

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