damon/links
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

多个A2C任务时防止同时同步引起错误

Browse Source
This commit is contained in:
root
2026-07-03 20:51:56 +08:00
parent 8508d51b66
commit 455b6669e8
7 changed files with 577 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
application/common/service/SplitNodeHealthService.php
+44 -14
View File
@@ -37,6 +37,8 @@ class SplitNodeHealthService
/**
* 当前是否适合向 Node 提交新 Browser 任务
*
* A2C 等 antiBot 任务走 real profile，应参考 queueReal 而非 standard 队列。
*/
public static function canAcceptWork(): bool
{
@@ -47,33 +49,61 @@ class SplitNodeHealthService
if (!empty($stats['shuttingDown'])) {
return false;
}
$queue = is_array($stats['queue'] ?? null) ? $stats['queue'] : [];
$config = is_array($stats['config'] ?? null) ? $stats['config'] : [];
$queued = (int) ($queue['queued'] ?? 0);
$active = (int) ($queue['active'] ?? 0);
$max = max(1, (int) ($config['maxConcurrentBrowsers'] ?? ($queue['max'] ?? 4)));
$snapshot = self::resolveQueueSnapshot($stats);
$threshold = SplitSyncConfigService::getNodeQueueSkipThreshold();
if ($queued >= $threshold) {
if ($snapshot['queued'] >= $threshold) {
return false;
}
return $active < $max;
return $snapshot['active'] < $snapshot['max'];
}
/**
* @return array{queued:int,active:int,max:int}
* @return array{queued:int,active:int,max:int,profile:string}
*/
public static function getQueueSnapshot(): array
{
$stats = self::fetchStats();
if ($stats === null) {
return ['queued' => 0, 'active' => 0, 'max' => 0];
return ['queued' => 0, 'active' => 0, 'max' => 0, 'profile' => 'real'];
}
$queue = is_array($stats['queue'] ?? null) ? $stats['queue'] : [];
$config = is_array($stats['config'] ?? null) ? $stats['config'] : [];
$snapshot = self::resolveQueueSnapshot($stats);
return [
'queued' => (int) ($queue['queued'] ?? 0),
'active' => (int) ($queue['active'] ?? 0),
'max' => max(1, (int) ($config['maxConcurrentBrowsers'] ?? ($queue['max'] ?? 4))),
'queued' => $snapshot['queued'],
'active' => $snapshot['active'],
'max' => $snapshot['max'],
'profile' => $snapshot['profile'],
];
}
/**
* 优先使用 real 队列（antiBot / A2C），无数据时回退 standard
*
* @param array<string, mixed> $stats
* @return array{queued:int,active:int,max:int,profile:string}
*/
private static function resolveQueueSnapshot(array $stats): array
{
$config = is_array($stats['config'] ?? null) ? $stats['config'] : [];
$queueReal = is_array($stats['queueReal'] ?? null) ? $stats['queueReal'] : [];
if ($queueReal !== []) {
return [
'queued' => (int) ($queueReal['queued'] ?? 0),
'active' => (int) ($queueReal['active'] ?? 0),
'max' => max(1, (int) ($config['maxConcurrentBrowsersReal'] ?? ($queueReal['max'] ?? 2))),
'profile' => 'real',
];
}
$queue = is_array($stats['queue'] ?? null) ? $stats['queue'] : [];
return [
'queued' => (int) ($queue['queued'] ?? 0),
'active' => (int) ($queue['active'] ?? 0),
'max' => max(1, (int) ($config['maxConcurrentBrowsers'] ?? ($queue['max'] ?? 4))),
'profile' => 'standard',
];
}
application/common/service/SplitTicketSyncDispatchService.php
+124 -12
View File
@@ -4,11 +4,15 @@ declare(strict_types=1);
namespace app\common\service;
use think\Db;
/**
* 工单手工同步 CLI 后台投递
*
* Web 请求仅负责鉴权与投递，实际 Spider 在独立 CLI 进程中执行，
* 避免长时间占用 PHP-FPM 与 Session 锁导致后台其它页面无法打开。
*
* Node 类型（A2C / Whatshub 等）在同一 nohup 子 shell 内串行执行，避免并行 Chrome 压垮温池。
*/
class SplitTicketSyncDispatchService
{
@@ -28,6 +32,13 @@ class SplitTicketSyncDispatchService
$think = $this->resolveThinkScript();
$root = $this->resolveRootPath();
$lockService = new SplitTicketSyncLockService();
$nodeTypeMap = array_flip(SplitScrmSpiderFactory::listNodeSyncTypes());
$typeMap = $this->loadTicketTypeMap($ticketIds);
/** @var int[] $nodeTicketIds */
$nodeTicketIds = [];
/** @var int[] $directTicketIds */
$directTicketIds = [];
foreach ($ticketIds as $rawId) {
$ticketId = (int) $rawId;
@@ -40,6 +51,15 @@ class SplitTicketSyncDispatchService
continue;
}
$ticketType = (string) ($typeMap[$ticketId] ?? '');
if ($ticketType !== '' && isset($nodeTypeMap[$ticketType])) {
$nodeTicketIds[] = $ticketId;
} else {
$directTicketIds[] = $ticketId;
}
}
foreach ($directTicketIds as $ticketId) {
if ($this->spawnCli($php, $think, $root, $ticketId)) {
$queued[] = $ticketId;
} else {
@@ -47,11 +67,26 @@ class SplitTicketSyncDispatchService
}
}
if ($nodeTicketIds !== []) {
$spawned = $this->spawnCliSequential($php, $think, $root, $nodeTicketIds);
if ($spawned) {
foreach ($nodeTicketIds as $ticketId) {
$queued[] = $ticketId;
}
} else {
foreach ($nodeTicketIds as $ticketId) {
$failed[] = $ticketId;
}
}
}
SplitTicketSyncLogger::log('web', 'manual sync dispatched', [
'queued' => $queued,
'skipped' => $skipped,
'failed' => $failed,
'phpCli' => $php,
'queued' => $queued,
'skipped' => $skipped,
'failed' => $failed,
'nodeSerial' => $nodeTicketIds,
'directParallel' => $directTicketIds,
'phpCli' => $php,
]);
return [
@@ -82,7 +117,37 @@ class SplitTicketSyncDispatchService
}
/**
* 后台启动 split:sync-tickets CLI
* @param int[] $ticketIds
* @return array<int, string> ticketId => ticket_type
*/
private function loadTicketTypeMap(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')
->select();
$map = [];
foreach ($rows as $row) {
$map[(int) $row['id']] = (string) ($row['ticket_type'] ?? '');
}
return $map;
}
/**
* 后台启动 split:sync-tickets CLI（单工单，并行投递）
*/
private function spawnCli(string $php, string $think, string $root, int $ticketId): bool
{
@@ -97,18 +162,63 @@ class SplitTicketSyncDispatchService
escapeshellarg($logFile)
);
return $this->runBackgroundCommand($root, $inner, $ticketId, 'parallel');
}
/**
* Node 类型工单：单条 nohup 内串行执行多个 CLI，避免并行 Chrome
*
* @param int[] $ticketIds
*/
private function spawnCliSequential(string $php, string $think, string $root, array $ticketIds): 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';
$parts[] = sprintf(
'%s %s split:sync-tickets --ticket=%d >> %s 2>&1',
escapeshellarg($php),
escapeshellarg($think),
$ticketId,
escapeshellarg($logFile)
);
}
if ($parts === []) {
return false;
}
$inner = implode('; ', $parts);
return $this->runBackgroundCommand($root, $inner, $ticketIds, '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 %s &', escapeshellarg($root), $inner);
$command = sprintf('cd %s && nohup bash -c %s &', escapeshellarg($root), escapeshellarg($inner));
}
if ($this->canUseExec()) {
@exec($command, $output, $exitCode);
SplitTicketSyncLogger::log('web', 'cli spawn exec', [
'ticketId' => $ticketId,
'command' => $command,
'exitCode' => $exitCode,
'ticketRef' => $ticketRef,
'mode' => $mode,
'command' => $command,
'exitCode' => $exitCode,
]);
return true;
}
@@ -116,14 +226,16 @@ class SplitTicketSyncDispatchService
if ($this->canUseShellExec()) {
@shell_exec($command);
SplitTicketSyncLogger::log('web', 'cli spawn shell_exec', [
'ticketId' => $ticketId,
'command' => $command,
'ticketRef' => $ticketRef,
'mode' => $mode,
'command' => $command,
]);
return true;
}
SplitTicketSyncLogger::log('web', 'cli spawn failed: exec disabled', [
'ticketId' => $ticketId,
'ticketRef' => $ticketRef,
'mode' => $mode,
]);
return false;