info('开始执行剧本资产生成任务调度...'); // 每次调度窗口执行一次历史 chunks 清理(任务完结后 chunks 仅供 SSE 推送,无长期保留价值) $cleaned = $this->cleanupExpiredChunks(); if ($cleaned > 0) { dLog('command')->info("已清理过期剧本资产生成chunks: {$cleaned} 条"); } $concurrency = (int)env('SCRIPT_TEXT_GENERATION_CONCURRENCY', 3); if ($concurrency < 1) { $concurrency = 1; } $start = time(); $idleSince = null; $spawnedTotal = 0; while (true) { $now = time(); // 孤儿任务恢复:processing 但超过5分钟且处理锁已空闲,说明 worker 异常退出,重置回 pending $this->recoverOrphanTasks(); // 连续 50 秒没有任何任务则退出 if ($idleSince !== null && $now - $idleSince >= 50) { dLog('command')->info('已连续50秒无待处理任务,退出本次调度'); break; } // 整体安全上限(防止异常情况下长时间占用) if ($now - $start > 75) { dLog('command')->info('达到调度窗口安全上限,退出本次调度'); break; } // 并发上限 = 配置上限 - 已在运行的 processing 任务数 - 本窗口已拉起数 $activeCount = (int)DB::table('mp_script_generate_tasks') ->where('status', 'processing') ->count(); $available = $concurrency - $activeCount - $spawnedTotal; if ($available <= 0) { dLog('command')->info("当前活跃任务已达并发上限,等待worker完成..."); sleep(5); continue; } $pendingIds = DB::table('mp_script_generate_tasks') ->where('status', 'pending') ->orderBy('id') ->pluck('id') ->all(); if (empty($pendingIds)) { if ($idleSince === null) { $idleSince = $now; } if ($now - $idleSince >= 50) { break; } sleep(5); continue; } $idleSince = null; foreach ($pendingIds as $taskId) { if ($available <= 0) { break; } // 仅当该任务的处理锁空闲时才拉起,避免与手动/其它调度重复 $lockName = TextScriptGenerateRunCommand::LOCK_PREFIX . $taskId; $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]); if (!$isFree || (int)$isFree->is_free !== 1) { dLog('command')->info("task_id: {$taskId} 正在被其他进程处理,跳过"); continue; } try { $this->spawnWorker((int)$taskId); $spawnedTotal++; $available--; dLog('command')->info("已拉起剧本资产生成worker, task_id: {$taskId}"); } catch (\Throwable $e) { dLog('command')->error("拉起剧本资产生成worker失败 task_id: {$taskId}: " . $e->getMessage()); logDB('command', 'error', '剧本资产生成拉起失败', [ 'task_id' => (int)$taskId, 'error' => $e->getMessage(), ]); } } sleep(5); } dLog('command')->info("剧本资产生成任务调度结束,本次拉起worker数: {$spawnedTotal}"); return 0; } /** * 清理已完结任务的过期 chunks * * 保留期默认 30 天,可通过环境变量 SCRIPT_GENERATE_CHUNK_RETENTION_DAYS 调整。 * 只清理 status 为 success/failed 且 completed_at 早于保留期的任务; * 按任务 id 升序游标分批删除,避免单次删除量过大造成长时间锁表。 * * @return int 删除的 chunks 记录数 */ private function cleanupExpiredChunks(): int { $retentionDays = (int)env('SCRIPT_GENERATE_CHUNK_RETENTION_DAYS', 7); if ($retentionDays < 1) { $retentionDays = 7; } $cutoff = date('Y-m-d H:i:s', strtotime("-{$retentionDays} days")); $batchRows = (int)env('SCRIPT_GENERATE_CHUNK_DELETE_BATCH', 5000); if ($batchRows < 1000) { $batchRows = 5000; } $deletedTotal = 0; // 按行数小批循环删除,避免单次 DELETE 量过大造成长事务/长锁。 // MySQL 单表 DELETE 支持 LIMIT;目标表是 chunks,子查询引用 tasks,不同表无同表限制。 $deleteSql = 'DELETE FROM mp_script_generate_task_chunks ' . 'WHERE task_id IN (' . 'SELECT id FROM mp_script_generate_tasks ' . 'WHERE status IN (?, ?) AND completed_at IS NOT NULL AND completed_at < ?' . ') LIMIT ' . (int)$batchRows; $maxRounds = 2000; // 防御性上限,正常数据远用不到 for ($round = 0; $round < $maxRounds; $round++) { $deleted = DB::affectingStatement($deleteSql, ['success', 'failed', $cutoff]); if ($deleted <= 0) { break; } $deletedTotal += $deleted; if ($deleted < $batchRows) { break; } } return $deletedTotal; } /** * 拉起一个后台 worker 子进程处理指定任务 * * @param int $taskId * @return void */ private function spawnWorker(int $taskId) { $command = [ $this->phpBinary(), base_path('artisan'), 'text:generate:run', (string)$taskId, ]; if (DIRECTORY_SEPARATOR === '/') { // Linux:用 nohup 后台启动并立即返回 $logFile = storage_path('logs/text-generate-worker-' . date('Ymd') . '.log'); $commandLine = 'nohup ' . implode(' ', array_map('escapeshellarg', $command)) . ' >> ' . escapeshellarg($logFile) . ' 2>&1 &'; exec($commandLine); dLog('command')->info('拉起剧本资产生成worker命令: ' . $commandLine); } else { // Windows:使用 cmd start /B 分离启动。 // 注意:不能用 Symfony Process::start(),其对象析构时会 stop(0) 终止刚拉起的子进程。 $logFile = storage_path('logs/text-generate-worker-' . date('Ymd') . '.log'); $commandLine = 'start /B "" ' . implode(' ', array_map('escapeshellarg', $command)) . ' >> ' . escapeshellarg($logFile) . ' 2>&1'; pclose(popen($commandLine, 'r')); dLog('command')->info('拉起剧本资产生成worker命令: ' . $commandLine); } } /** * 恢复异常退出遗留的孤儿任务 * * worker 进程被强杀/崩溃时不会执行收尾,任务会一直停留在 processing。 * 若任务已 processing 超过 5 分钟且其处理锁处于空闲状态,说明原 worker 已不在, * 将任务重置回 pending,等待重新调度。 * * @return int 恢复的任务数 */ private function recoverOrphanTasks(): int { $recovered = 0; $threshold = date('Y-m-d H:i:s', strtotime('-5 minutes')); $candidates = DB::table('mp_script_generate_tasks') ->where('status', 'processing') ->where('started_at', '<', $threshold) ->pluck('id') ->all(); foreach ($candidates as $taskId) { $lockName = TextScriptGenerateRunCommand::LOCK_PREFIX . $taskId; $isFree = DB::selectOne('SELECT IS_FREE_LOCK(?) AS is_free', [$lockName]); if (!$isFree || (int)$isFree->is_free !== 1) { continue; } $updated = DB::table('mp_script_generate_tasks') ->where('id', $taskId) ->where('status', 'processing') ->update([ 'status' => 'pending', 'error_message' => null, 'completed_at' => null, 'updated_at' => date('Y-m-d H:i:s'), ]); if ($updated) { $recovered++; dLog('command')->warning("孤儿任务恢复, task_id: {$taskId} 已重置为pending"); logDB('command', 'warning', '孤儿任务恢复', [ 'task_id' => (int)$taskId, 'message' => 'worker异常退出,任务重置为pending等待重新调度', ]); } } return $recovered; } /** * 获取当前 PHP 可执行文件的真实路径 * * @return string */ private function phpBinary() { $envBinary = getenv('PHP_BINARY'); if ($envBinary && is_executable($envBinary)) { return $envBinary; } if (DIRECTORY_SEPARATOR === '/' && ($real = @readlink('/proc/self/exe')) && is_executable($real)) { return $real; } return PHP_BINARY; } }