loadTicketMetaList($ticketIds); /** @var list $toDispatch */ $toDispatch = []; $skipped = []; $failed = []; foreach ($metaList as $meta) { $ticketId = (int) $meta['id']; if ((int) ($meta['manual_manage'] ?? 0) === 1) { $skipped[] = $ticketId; continue; } if ($lockService->isLocked($ticketId)) { $skipped[] = $ticketId; continue; } $toDispatch[] = $meta; } $dispatchResult = $this->dispatchTicketMetaList($toDispatch, 'manual'); SplitTicketSyncLogger::log('web', 'manual sync dispatched', [ 'queued' => $dispatchResult['queued'], 'skipped' => $skipped, 'failed' => $dispatchResult['failed'], 'phpCli' => $this->resolvePhpBinary(), ]); return [ 'queued' => $dispatchResult['queued'], 'skipped' => $skipped, 'failed' => $dispatchResult['failed'], ]; } /** * Cron 自动同步:将挑选后的到期工单投递到后台 CLI(Cron 进程立即返回) * * @param list $tickets * @return array{ * queued:list>, * skipped:list>, * failed:list>, * stoppedByNodeBusy:bool * } */ public function dispatchAuto(array $tickets): array { $lockService = new SplitTicketSyncLockService(); /** @var list $toDispatch */ $toDispatch = []; /** @var list> $skipped */ $skipped = []; foreach ($tickets as $ticket) { $meta = $this->normalizeTicketMeta($ticket); if ($meta === null) { continue; } $ticketId = (int) $meta['id']; if ($lockService->isLocked($ticketId)) { $skipped[] = $this->buildDispatchEntry( $ticketId, (string) $meta['ticket_type'], (string) $meta['ticket_url'], '工单正在同步中' ); continue; } $toDispatch[] = $meta; } $dispatchResult = $this->dispatchTicketMetaList($toDispatch, 'auto'); SplitTicketSyncLogger::log('cron', 'auto sync dispatched', [ 'queuedCount' => count($dispatchResult['queued']), 'skippedCount' => count($skipped) + count($dispatchResult['skipped']), 'failedCount' => count($dispatchResult['failed']), 'stoppedByNodeBusy' => $dispatchResult['stoppedByNodeBusy'], 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), ]); return [ 'queued' => $dispatchResult['queued'], 'skipped' => array_merge($skipped, $dispatchResult['skipped']), 'failed' => $dispatchResult['failed'], 'stoppedByNodeBusy' => $dispatchResult['stoppedByNodeBusy'], ]; } /** * 查询仍在同步中的工单 ID(依据 runtime 文件锁) * * @param int[] $ticketIds * @return int[] */ public function filterSyncingIds(array $ticketIds): array { $lockService = new SplitTicketSyncLockService(); $syncing = []; foreach ($ticketIds as $rawId) { $ticketId = (int) $rawId; if ($ticketId > 0 && $lockService->isLocked($ticketId)) { $syncing[] = $ticketId; } } return $syncing; } /** * @param list $metaList * @return array{ * queued:list>, * skipped:list>, * failed:list>, * stoppedByNodeBusy:bool * } */ private function dispatchTicketMetaList(array $metaList, string $source): array { $php = $this->resolvePhpBinary(); $think = $this->resolveThinkScript(); $root = $this->resolveRootPath(); /** @var list> $queued */ $queued = []; /** @var list> $skipped */ $skipped = []; /** @var list> $failed */ $failed = []; $stoppedByNodeBusy = false; /** @var list $apiFirstList */ $apiFirstList = []; /** @var list $directList */ $directList = []; /** @var array> $nodeGroups */ $nodeGroups = []; foreach ($metaList as $meta) { $ticketType = (string) $meta['ticket_type']; if (SplitScrmSpiderFactory::isApiFirstSyncType($ticketType)) { $apiFirstList[] = $meta; continue; } if (!SplitScrmSpiderFactory::requiresNode($ticketType)) { $directList[] = $meta; continue; } $sessionKey = AntiBotConfigBuilder::resolveSessionKey( $ticketType, (string) $meta['ticket_url'] ); $nodeGroups[$sessionKey][] = $meta; } foreach ($apiFirstList as $meta) { $this->spawnSingleMeta($php, $think, $root, $meta, $source, $queued, $failed); } foreach ($directList as $meta) { $this->spawnSingleMeta($php, $think, $root, $meta, $source, $queued, $failed); } foreach ($nodeGroups as $sessionKey => $group) { if ($group === []) { continue; } /** @var list $nodeReady */ $nodeReady = []; foreach ($group as $meta) { $ticketType = (string) $meta['ticket_type']; if (!SplitNodeHealthService::canAcceptWorkForTicketType($ticketType)) { $stoppedByNodeBusy = true; $skipped[] = $this->buildDispatchEntry( (int) $meta['id'], $ticketType, (string) $meta['ticket_url'], 'Node 队列繁忙,本轮未投递' ); continue; } $nodeReady[] = $meta; } if ($nodeReady === []) { continue; } if (count($nodeReady) === 1) { $this->spawnSingleMeta($php, $think, $root, $nodeReady[0], $source, $queued, $failed); continue; } $ids = array_map(static function (array $row): int { return (int) $row['id']; }, $nodeReady); if ($this->spawnCliSequential($php, $think, $root, $ids, $source)) { foreach ($nodeReady as $meta) { $queued[] = $this->buildQueuedEntry($meta, $sessionKey, 'node-group-serial'); } } else { foreach ($nodeReady as $meta) { $failed[] = $this->buildDispatchEntry( (int) $meta['id'], (string) $meta['ticket_type'], (string) $meta['ticket_url'], 'CLI 投递失败' ); } } } return [ 'queued' => $queued, 'skipped' => $skipped, 'failed' => $failed, 'stoppedByNodeBusy' => $stoppedByNodeBusy, ]; } /** * @param list> $queued * @param list> $failed * @param array{id:int,ticket_type:string,ticket_url:string} $meta */ private function spawnSingleMeta( string $php, string $think, string $root, array $meta, string $source, array &$queued, array &$failed ): void { $ticketId = (int) $meta['id']; if ($this->spawnCli($php, $think, $root, $ticketId, $source)) { $queued[] = $this->buildQueuedEntry($meta, '', 'parallel'); return; } $failed[] = $this->buildDispatchEntry( $ticketId, (string) $meta['ticket_type'], (string) $meta['ticket_url'], 'CLI 投递失败' ); } /** * @param array{id:int,ticket_type:string,ticket_url:string} $meta * @return array */ private function buildQueuedEntry(array $meta, string $sessionKey, string $mode): array { return [ 'ticketId' => (int) $meta['id'], 'ticketType' => (string) $meta['ticket_type'], 'ticketUrl' => (string) $meta['ticket_url'], 'sessionKey' => $sessionKey, 'mode' => $mode, 'dispatched' => true, ]; } /** * @return array{ticketId:int,ticketType:string,ticketUrl:string,reason:string} */ private function buildDispatchEntry(int $ticketId, string $ticketType, string $ticketUrl, string $reason): array { return [ 'ticketId' => $ticketId, 'ticketType' => $ticketType, 'ticketUrl' => $ticketUrl, 'reason' => $reason, ]; } /** * @param int[] $ticketIds * @return list */ private function loadTicketMetaList(array $ticketIds): array { $ids = []; foreach ($ticketIds as $rawId) { $ticketId = (int) $rawId; if ($ticketId > 0) { $ids[] = $ticketId; } } if ($ids === []) { return []; } $rows = Db::name('split_ticket') ->where('id', 'in', $ids) ->field('id,ticket_type,ticket_url,manual_manage') ->select(); $list = []; foreach ($rows as $row) { $list[] = [ 'id' => (int) $row['id'], 'ticket_type' => (string) ($row['ticket_type'] ?? ''), 'ticket_url' => (string) ($row['ticket_url'] ?? ''), 'manual_manage' => (int) ($row['manual_manage'] ?? 0), ]; } return $list; } /** * @param array|object $ticket * @return array{id:int,ticket_type:string,ticket_url:string}|null */ private function normalizeTicketMeta($ticket): ?array { if (is_object($ticket) && method_exists($ticket, 'toArray')) { $ticket = $ticket->toArray(); } if (!is_array($ticket)) { return null; } $ticketId = (int) ($ticket['id'] ?? 0); if ($ticketId <= 0) { return null; } return [ 'id' => $ticketId, 'ticket_type' => (string) ($ticket['ticket_type'] ?? ''), 'ticket_url' => (string) ($ticket['ticket_url'] ?? ''), ]; } /** * 后台启动 split:sync-tickets CLI(单工单,独立进程) */ private function spawnCli(string $php, string $think, string $root, int $ticketId, string $source): bool { $logDir = $this->resolveLogDir(); $logFile = $logDir . 'cli_' . $ticketId . '_' . date('YmdHis') . '.log'; $syncMode = $this->normalizeSyncModeForCli($source); $inner = sprintf( '%s %s split:sync-tickets --ticket=%d --mode=%s >> %s 2>&1', escapeshellarg($php), escapeshellarg($think), $ticketId, $syncMode, escapeshellarg($logFile) ); return $this->runBackgroundCommand($root, $inner, $ticketId, $source . '-parallel'); } /** * 同一 sessionKey 内串行执行多个 CLI(单 nohup 子 shell) * * @param int[] $ticketIds */ private function spawnCliSequential(string $php, string $think, string $root, array $ticketIds, string $source): bool { if ($ticketIds === []) { return false; } $logDir = $this->resolveLogDir(); $parts = []; foreach ($ticketIds as $ticketId) { $ticketId = (int) $ticketId; if ($ticketId <= 0) { continue; } $logFile = $logDir . 'cli_' . $ticketId . '_' . date('YmdHis') . '.log'; $syncMode = $this->normalizeSyncModeForCli($source); $parts[] = sprintf( '%s %s split:sync-tickets --ticket=%d --mode=%s >> %s 2>&1', escapeshellarg($php), escapeshellarg($think), $ticketId, $syncMode, escapeshellarg($logFile) ); } if ($parts === []) { return false; } $inner = implode('; ', $parts); return $this->runBackgroundCommand($root, $inner, $ticketIds, $source . '-node-serial'); } /** * @param int|int[] $ticketRef 日志用 ticketId 或 ID 列表 */ private function runBackgroundCommand(string $root, string $inner, $ticketRef, string $mode): bool { if ($this->isWindows()) { $command = sprintf('start /B cmd /C %s', $inner); } else { $command = sprintf('cd %s && nohup bash -c %s &', escapeshellarg($root), escapeshellarg($inner)); } if ($this->canUseExec()) { @exec($command, $output, $exitCode); SplitTicketSyncLogger::log('dispatch', 'cli spawn exec', [ 'ticketRef' => $ticketRef, 'mode' => $mode, 'command' => $command, 'exitCode' => $exitCode, ]); return true; } if ($this->canUseShellExec()) { @shell_exec($command); SplitTicketSyncLogger::log('dispatch', 'cli spawn shell_exec', [ 'ticketRef' => $ticketRef, 'mode' => $mode, 'command' => $command, ]); return true; } SplitTicketSyncLogger::log('dispatch', 'cli spawn failed: exec disabled', [ 'ticketRef' => $ticketRef, 'mode' => $mode, ]); return false; } private function resolvePhpBinary(): string { if (method_exists(SplitSyncConfigService::class, 'getCliPhpBinary')) { return SplitSyncConfigService::getCliPhpBinary(); } return $this->resolvePhpBinaryFallback(); } /** * 当 SplitSyncConfigService 未部署 getCliPhpBinary 时的兜底解析 */ private function resolvePhpBinaryFallback(): string { if (defined('PHP_BINDIR') && PHP_BINDIR !== '') { $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php'; if ($this->isUsableCliPhp($candidate)) { return $candidate; } } foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) { if ($this->isUsableCliPhp($candidate)) { return $candidate; } } return 'php'; } private function isUsableCliPhp(string $path): bool { if (stripos($path, 'fpm') !== false) { return false; } if ($path === 'php') { return true; } return is_file($path) && is_executable($path); } private function resolveThinkScript(): string { $root = $this->resolveRootPath(); return rtrim($root, '/\\') . DIRECTORY_SEPARATOR . 'think'; } private function resolveRootPath(): string { if (defined('ROOT_PATH') && ROOT_PATH !== '') { return rtrim(ROOT_PATH, '/\\') . DIRECTORY_SEPARATOR; } return rtrim(dirname(__DIR__, 3), '/\\') . DIRECTORY_SEPARATOR; } private function resolveLogDir(): string { $runtime = defined('RUNTIME_PATH') ? RUNTIME_PATH : ($this->resolveRootPath() . 'runtime' . DIRECTORY_SEPARATOR); $dir = rtrim($runtime, '/\\') . DIRECTORY_SEPARATOR . 'split_ticket_sync' . DIRECTORY_SEPARATOR; if (!is_dir($dir)) { @mkdir($dir, 0755, true); } return $dir; } private function isWindows(): bool { return strtoupper(substr(PHP_OS, 0, 3)) === 'WIN'; } /** * @return string[] */ private function disabledFunctions(): array { $raw = (string) ini_get('disable_functions'); if ($raw === '') { return []; } return array_filter(array_map('trim', explode(',', $raw))); } private function canUseExec(): bool { return function_exists('exec') && !in_array('exec', $this->disabledFunctions(), true); } private function canUseShellExec(): bool { return function_exists('shell_exec') && !in_array('shell_exec', $this->disabledFunctions(), true); } /** * 投递 CLI 时携带 --mode,供 sync_success_mode 区分自动/手动 */ private function normalizeSyncModeForCli(string $source): string { return in_array($source, ['auto', 'manual'], true) ? $source : 'manual'; } }