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(); } }