102 lines
3.6 KiB
PHP
102 lines
3.6 KiB
PHP
<?php
|
|
|
|
namespace App\Console\Commands;
|
|
|
|
use Illuminate\Console\Command;
|
|
use Illuminate\Support\Facades\Log;
|
|
use InvalidArgumentException;
|
|
use Modules\EcommerceFinance\Services\UpstreamOdbcConnection;
|
|
use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
|
|
use Modules\EcommerceFinance\Services\UpstreamSnapshotRefresherRegistry;
|
|
use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
|
|
use Throwable;
|
|
|
|
class UpstreamRefreshActiveSnapshots extends Command
|
|
{
|
|
protected $signature = 'upstream:snapshots:refresh
|
|
{source : config/upstreams.php 中登记的数据源标识}
|
|
{--limit= : 本轮最多刷新的活跃查询数}';
|
|
|
|
protected $description = '在各自时间、并发和查询预算内刷新指定上游的活跃快照';
|
|
|
|
public function handle(
|
|
UpstreamSourceRegistry $sources,
|
|
UpstreamQueryCacheFactory $caches,
|
|
UpstreamSnapshotRefresherRegistry $refreshers,
|
|
): int {
|
|
$source = (string) $this->argument('source');
|
|
|
|
try {
|
|
$definition = $sources->definition($source);
|
|
} catch (InvalidArgumentException $exception) {
|
|
$this->components->error($exception->getMessage());
|
|
|
|
return self::FAILURE;
|
|
}
|
|
|
|
if (! $sources->enabled($source)) {
|
|
$this->components->info($definition['label'] . '同步未启用，本轮跳过。');
|
|
|
|
return self::SUCCESS;
|
|
}
|
|
|
|
try {
|
|
$cache = $caches->forSource($source);
|
|
$refresher = $refreshers->forSource($source);
|
|
} catch (InvalidArgumentException $exception) {
|
|
$this->components->error($exception->getMessage());
|
|
|
|
return self::FAILURE;
|
|
}
|
|
|
|
$configuredLimit = max((int) $cache->setting('snapshots.refresh_limit', 2), 1);
|
|
$requestedLimit = (int) ($this->option('limit') ?: $configuredLimit);
|
|
$entries = $cache->activeEntries(min(max($requestedLimit, 1), $configuredLimit));
|
|
$timeBudget = max((int) $cache->setting('snapshots.refresh_time_budget_seconds', 45), 10);
|
|
$startedAt = microtime(true);
|
|
|
|
if ($entries === []) {
|
|
$this->components->info('没有需要刷新的活跃查询。');
|
|
|
|
return self::SUCCESS;
|
|
}
|
|
|
|
$fresh = 0;
|
|
$stale = 0;
|
|
|
|
foreach ($entries as $entry) {
|
|
if ((microtime(true) - $startedAt) >= $timeBudget) {
|
|
$this->components->info('本轮刷新已达到时间预算，剩余查询留到下一轮。');
|
|
break;
|
|
}
|
|
|
|
try {
|
|
$status = $refresher->refresh($entry);
|
|
$status === 'fresh' ? $fresh++ : $stale++;
|
|
} catch (Throwable $exception) {
|
|
// 每个来源独立停止和重置，不影响其他来源的连接与刷新队列。
|
|
UpstreamOdbcConnection::reset($source);
|
|
Log::error('上游活跃快照刷新失败，本来源本轮已停止。', [
|
|
'source' => $definition['key'],
|
|
'source_label' => $definition['label'],
|
|
'scope' => $entry['scope'] ?? 'unknown',
|
|
'exception' => $exception::class,
|
|
]);
|
|
$this->components->error('上游刷新失败，本轮已停止；错误类型：' . $exception::class);
|
|
|
|
return self::FAILURE;
|
|
}
|
|
|
|
$pauseMs = max((int) $cache->setting('snapshots.refresh_pause_ms', 300), 0);
|
|
|
|
if ($pauseMs > 0) {
|
|
usleep($pauseMs * 1000);
|
|
}
|
|
}
|
|
|
|
$this->components->info("快照刷新完成：新鲜 {$fresh}，沿用旧快照 {$stale}。");
|
|
|
|
return self::SUCCESS;
|
|
}
|
|
}
|