| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115 |
- <?php
- namespace App\Console\Commands;
- use Illuminate\Console\Command;
- use Illuminate\Support\Facades\DB;
- use Symfony\Component\Process\Process;
- /**
- * 批量生成分集定时任务(调度器)
- * 每次执行:找出有 pending 任务的 anime_id,为最多 N 个(默认5)空闲 anime 各拉起一个后台子进程,
- * 由子进程独占处理对应 anime 的待生成剧集。
- */
- class ProcessBatchEpisodeGenerationCommand extends Command
- {
- /**
- * The name and signature of the console command.
- *
- * @var string
- */
- protected $signature = 'batch:generate-episodes';
- /**
- * The console command description.
- *
- * @var string
- */
- protected $description = '批量生成分集定时任务:并发调度不同anime_id的子进程处理,并发数默认5';
- /**
- * Execute the console command.
- *
- * @return int
- */
- public function handle()
- {
- dLog('command')->info('开始执行批量生成分集调度任务...');
- $concurrency = (int) env('BATCH_EPISODE_GENERATION_CONCURRENCY', 5);
- if ($concurrency < 1) {
- $concurrency = 1;
- }
- // 找出所有有待处理任务的 anime_id
- $animeIds = DB::table('mp_batch_episode_generation_details')
- ->where('status', 'pending')
- ->distinct()
- ->pluck('anime_id')
- ->map(function ($animeId) {
- return (int) $animeId;
- })
- ->sort()
- ->values()
- ->toArray();
- if (empty($animeIds)) {
- dLog('command')->info('没有待处理的任务,本次调度结束');
- return 0;
- }
- $spawned = 0;
- foreach ($animeIds as $animeId) {
- if ($spawned >= $concurrency) {
- dLog('command')->info("已达到最大并发数 {$concurrency},停止调度");
- break;
- }
- $lockName = ProcessBatchAnimeEpisodeGenerationCommand::LOCK_PREFIX . $animeId;
- // 仅当该 anime 的处理锁空闲时才拉起子进程;
- // 真正的加锁由子进程完成,避免锁挂在调度进程的连接上
- $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]);
- if (!$isFree || (int) $isFree->is_free !== 1) {
- dLog('command')->info("anime_id: {$animeId} 正在被其他进程处理,跳过");
- continue;
- }
- try {
- $this->spawnWorker($animeId);
- $spawned++;
- dLog('command')->info("已拉起子进程处理 anime_id: {$animeId}");
- } catch (\Throwable $e) {
- dLog('command')->error("拉起子进程失败 anime_id: {$animeId}: " . $e->getMessage());
- logDB('command', 'error', '批量分集调度:拉起子进程失败', [
- 'anime_id' => $animeId,
- 'error' => $e->getMessage(),
- 'trace' => $e->getTraceAsString()
- ]);
- }
- }
- dLog('command')->info("批量生成分集调度完成,本次拉起子进程数: {$spawned}");
- return 0;
- }
- /**
- * 拉起一个后台子进程,独占处理指定 anime_id 的待生成剧集
- *
- * @param int $animeId
- * @return void
- */
- private function spawnWorker($animeId)
- {
- $process = new Process([
- PHP_BINARY,
- base_path('artisan'),
- 'batch:generate-episodes:anime',
- (string) $animeId,
- ]);
- $process->setTimeout(null);
- $process->setIdleTimeout(null);
- $process->disableOutput();
- $process->start();
- }
- }
|