502 lines
17 KiB
PHP
502 lines
17 KiB
PHP
<?php
|
||
|
||
namespace Modules\EcommerceFinance\Services;
|
||
|
||
use Closure;
|
||
use Illuminate\Contracts\Cache\Lock;
|
||
use Illuminate\Contracts\Cache\LockTimeoutException;
|
||
use Illuminate\Support\Facades\Cache;
|
||
use Illuminate\Support\Facades\Log;
|
||
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 上游查询的本地快照、请求合并和压力保护。
|
||
*
|
||
* 页面命中新鲜快照时立即返回；命中过期快照时也立即返回，并把刷新登记给
|
||
* 后台任务。只有完全没有本地快照的冷查询才会在请求内受控访问一次上游。
|
||
*/
|
||
class JianSunQueryCache
|
||
{
|
||
private const CACHE_VERSION = 'v2';
|
||
|
||
private const REGISTRY_KEY = 'jiansun:snapshots:active:v2';
|
||
|
||
private const GLOBAL_CIRCUIT_KEY = 'jiansun:upstream-circuit:v2';
|
||
|
||
private ?array $lastReadState = null;
|
||
|
||
/**
|
||
* 读取本地快照；过期快照采用 stale-while-revalidate，不阻塞页面。
|
||
*/
|
||
public function remember(string $scope, array $parameters, Closure $resolver): mixed
|
||
{
|
||
$parameters = $this->normalize($parameters);
|
||
$key = $this->cacheKey($scope, $parameters);
|
||
$cached = Cache::get($key);
|
||
|
||
if ($this->hasValue($cached)) {
|
||
$this->register($scope, $parameters, false);
|
||
$this->setReadState('fresh', $cached);
|
||
|
||
return $cached['value'];
|
||
}
|
||
|
||
$stale = Cache::get($key . ':stale');
|
||
|
||
if ($this->hasValue($stale)) {
|
||
// 页面不等待上游；下一轮后台任务优先刷新这个查询。
|
||
$this->register($scope, $parameters, true);
|
||
$this->setReadState('stale', $stale);
|
||
|
||
return $stale['value'];
|
||
}
|
||
|
||
if (! $this->isScopeAllowed($scope)) {
|
||
throw new ServiceUnavailableHttpException(30, '该数据范围仍在低压力验证中，请稍后再试。');
|
||
}
|
||
|
||
// 首次访问没有任何本地数据时才允许同步获取一次；先登记，失败后后台
|
||
// 仍可在熔断结束后继续补齐快照。
|
||
$this->register($scope, $parameters, true);
|
||
$resolved = $this->resolve($scope, $parameters, $resolver, false);
|
||
$payload = Cache::get($key) ?: Cache::get($key . ':stale');
|
||
$this->setReadState($resolved['status'], is_array($payload) ? $payload : []);
|
||
|
||
return $resolved['value'];
|
||
}
|
||
|
||
/**
|
||
* 供 API 响应暴露数据新鲜度，不改变既有 JSON 结构。
|
||
*/
|
||
public function responseHeaders(): array
|
||
{
|
||
if ($this->lastReadState === null) {
|
||
return [];
|
||
}
|
||
|
||
return array_filter([
|
||
'X-Caiwu-Data-Status' => $this->lastReadState['status'] ?? null,
|
||
'X-Caiwu-Data-Synced-At' => $this->lastReadState['cached_at'] ?? null,
|
||
]);
|
||
}
|
||
|
||
/**
|
||
* 后台刷新一个最近使用过的快照。
|
||
*
|
||
* @return string fresh|stale
|
||
*/
|
||
public function refresh(string $scope, array $parameters, Closure $resolver): string
|
||
{
|
||
$parameters = $this->normalize($parameters);
|
||
|
||
if (! $this->isScopeAllowed($scope)) {
|
||
return 'stale';
|
||
}
|
||
|
||
return $this->resolve($scope, $parameters, $resolver, true)['status'];
|
||
}
|
||
|
||
/**
|
||
* 返回最近实际访问且已经到期的查询，紧急项和列表优先。
|
||
*/
|
||
public function activeEntries(?int $limit = null): array
|
||
{
|
||
$now = time();
|
||
$activeTtl = max((int) config('jiansun.snapshots.active_ttl_seconds', 21600), 900);
|
||
$entries = Cache::get(self::REGISTRY_KEY, []);
|
||
|
||
if (! is_array($entries)) {
|
||
return [];
|
||
}
|
||
|
||
$entries = array_filter($entries, function (mixed $entry) use ($now, $activeTtl): bool {
|
||
return is_array($entry)
|
||
&& isset($entry['seen_at'], $entry['scope'], $entry['parameters'])
|
||
&& $this->isScopeAllowed((string) $entry['scope'])
|
||
&& ($now - (int) $entry['seen_at']) <= $activeTtl
|
||
&& $this->isDue($entry, $now);
|
||
});
|
||
|
||
usort($entries, function (array $a, array $b): int {
|
||
$urgent = ((int) ($b['urgent'] ?? 0)) <=> ((int) ($a['urgent'] ?? 0));
|
||
|
||
if ($urgent !== 0) {
|
||
return $urgent;
|
||
}
|
||
|
||
$priority = $this->priorityFor((string) $b['scope']) <=> $this->priorityFor((string) $a['scope']);
|
||
|
||
if ($priority !== 0) {
|
||
return $priority;
|
||
}
|
||
|
||
return ((int) $b['seen_at']) <=> ((int) $a['seen_at']);
|
||
});
|
||
|
||
$limit ??= max((int) config('jiansun.snapshots.refresh_limit', 2), 1);
|
||
|
||
return array_slice($entries, 0, $limit);
|
||
}
|
||
|
||
private function resolve(string $scope, array $parameters, Closure $resolver, bool $force): array
|
||
{
|
||
$key = $this->cacheKey($scope, $parameters);
|
||
$staleKey = $key . ':stale';
|
||
$queryId = hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE));
|
||
$queryLock = Cache::lock(
|
||
'jiansun:query-lock:' . $queryId,
|
||
max((int) config('jiansun.snapshots.lock_seconds', 45), 15)
|
||
);
|
||
|
||
try {
|
||
return $queryLock->block(
|
||
max((int) config('jiansun.snapshots.lock_wait_seconds', 3), 1),
|
||
function () use ($scope, $parameters, $resolver, $force, $key, $staleKey, $queryId): array {
|
||
if (! $force) {
|
||
$cached = Cache::get($key);
|
||
|
||
if ($this->hasValue($cached)) {
|
||
$this->markRefreshed($scope, $parameters);
|
||
|
||
return ['value' => $cached['value'], 'status' => 'fresh'];
|
||
}
|
||
}
|
||
|
||
if ($this->isCircuitOpen() || Cache::has($this->queryCooldownKey($queryId))) {
|
||
return $this->staleOrFail($staleKey, '上游数据正在安全冷却，请稍后再试。');
|
||
}
|
||
|
||
if (! $this->consumeBudget()) {
|
||
return $this->staleOrFail($staleKey, '上游查询刷新已达到安全上限，请稍后再试。');
|
||
}
|
||
|
||
$slot = $this->acquireUpstreamSlot();
|
||
|
||
if ($slot === null) {
|
||
return $this->staleOrFail($staleKey, '上游同步任务繁忙，请稍后再试。');
|
||
}
|
||
|
||
$startedAt = microtime(true);
|
||
|
||
try {
|
||
$value = $resolver();
|
||
} catch (Throwable $exception) {
|
||
$this->recordFailure($scope, $queryId, $exception);
|
||
|
||
return $this->staleOrFail($staleKey, '上游数据暂时不可用，请稍后再试。', $exception);
|
||
} finally {
|
||
$slot->release();
|
||
}
|
||
|
||
$durationMs = (int) round((microtime(true) - $startedAt) * 1000);
|
||
$payload = [
|
||
'value' => $value,
|
||
'cached_at' => now()->toIso8601String(),
|
||
'cached_at_ts' => time(),
|
||
];
|
||
|
||
Cache::put($key, $payload, max($this->ttlFor($scope), 60));
|
||
Cache::put(
|
||
$staleKey,
|
||
$payload,
|
||
max((int) config('jiansun.snapshots.stale_ttl_seconds', 86400), $this->ttlFor($scope))
|
||
);
|
||
|
||
$this->recordSuccess($scope, $parameters, $queryId, $durationMs, $value);
|
||
|
||
return ['value' => $value, 'status' => 'fresh'];
|
||
}
|
||
);
|
||
} catch (LockTimeoutException $exception) {
|
||
return $this->staleOrFail($staleKey, '相同数据正在同步，请稍后再试。', $exception);
|
||
}
|
||
}
|
||
|
||
private function acquireUpstreamSlot(): ?Lock
|
||
{
|
||
$slotCount = min(max((int) config('jiansun.snapshots.max_concurrency', 2), 1), 4);
|
||
$waitMs = min(max((int) config('jiansun.snapshots.slot_wait_ms', 1500), 0), 5000);
|
||
$deadline = microtime(true) + ($waitMs / 1000);
|
||
$lockSeconds = max(
|
||
(int) config('jiansun.connection.connect_timeout', 5)
|
||
+ (int) config('jiansun.connection.query_timeout', 15)
|
||
+ 10,
|
||
15
|
||
);
|
||
|
||
do {
|
||
for ($slot = 0; $slot < $slotCount; $slot++) {
|
||
$lock = Cache::lock('jiansun:upstream-slot:' . $slot, $lockSeconds);
|
||
|
||
if ($lock->get()) {
|
||
return $lock;
|
||
}
|
||
}
|
||
|
||
if (microtime(true) >= $deadline) {
|
||
break;
|
||
}
|
||
|
||
usleep(50000);
|
||
} while (true);
|
||
|
||
return null;
|
||
}
|
||
|
||
private function consumeBudget(): bool
|
||
{
|
||
$window = max((int) config('jiansun.snapshots.budget_window_seconds', 60), 60);
|
||
$limit = max((int) config('jiansun.snapshots.miss_budget_per_window', 6), 1);
|
||
$bucket = intdiv(time(), $window);
|
||
$key = 'jiansun:upstream-budget:' . $bucket;
|
||
$lock = Cache::lock('jiansun:upstream-budget-lock', 3);
|
||
|
||
try {
|
||
return $lock->block(1, function () use ($key, $limit, $window): bool {
|
||
$count = (int) Cache::get($key, 0);
|
||
|
||
if ($count >= $limit) {
|
||
return false;
|
||
}
|
||
|
||
Cache::put($key, $count + 1, $window + 60);
|
||
|
||
return true;
|
||
});
|
||
} catch (LockTimeoutException) {
|
||
// 预算状态不确定时宁可少查一次，也不冒险放大上游流量。
|
||
return false;
|
||
}
|
||
}
|
||
|
||
private function recordFailure(string $scope, string $queryId, Throwable $exception): void
|
||
{
|
||
$queryCooldown = max((int) config('jiansun.snapshots.query_failure_cooldown_seconds', 300), 60);
|
||
Cache::put($this->queryCooldownKey($queryId), true, $queryCooldown);
|
||
|
||
if ($exception instanceof ServiceUnavailableHttpException) {
|
||
$globalCooldown = max((int) config('jiansun.snapshots.failure_cooldown_seconds', 300), 60);
|
||
Cache::put(self::GLOBAL_CIRCUIT_KEY, true, $globalCooldown);
|
||
}
|
||
|
||
Log::warning('JianSun 上游查询失败，已进入冷却。', [
|
||
'scope' => $scope,
|
||
'exception' => $exception::class,
|
||
]);
|
||
}
|
||
|
||
private function recordSuccess(
|
||
string $scope,
|
||
array $parameters,
|
||
string $queryId,
|
||
int $durationMs,
|
||
mixed $value
|
||
): void {
|
||
$slowMs = max((int) config('jiansun.snapshots.slow_query_ms', 5000), 1000);
|
||
|
||
if ($durationMs >= $slowMs) {
|
||
$cooldown = max((int) config('jiansun.snapshots.slow_query_cooldown_seconds', 300), 60);
|
||
Cache::put($this->queryCooldownKey($queryId), true, $cooldown);
|
||
Log::warning('JianSun 上游查询较慢，已降低该查询刷新频率。', [
|
||
'scope' => $scope,
|
||
'duration_ms' => $durationMs,
|
||
]);
|
||
}
|
||
|
||
$this->markRefreshed($scope, $parameters, [
|
||
'last_duration_ms' => $durationMs,
|
||
'last_row_count' => $this->rowCount($value),
|
||
'last_refreshed_at' => time(),
|
||
]);
|
||
}
|
||
|
||
private function register(string $scope, array $parameters, bool $urgent): void
|
||
{
|
||
$entryKey = $this->entryKey($scope, $parameters);
|
||
$touchSeconds = $urgent ? 30 : 300;
|
||
|
||
if (! Cache::add('jiansun:registry-touch:' . $entryKey . ':' . (int) $urgent, true, $touchSeconds)) {
|
||
return;
|
||
}
|
||
|
||
$this->updateRegistry(function (array $entries) use ($entryKey, $scope, $parameters, $urgent): array {
|
||
$existing = (array) ($entries[$entryKey] ?? []);
|
||
$entries[$entryKey] = array_merge($existing, [
|
||
'scope' => $scope,
|
||
'parameters' => $parameters,
|
||
'seen_at' => time(),
|
||
'urgent' => $urgent || (bool) ($existing['urgent'] ?? false),
|
||
]);
|
||
|
||
return $entries;
|
||
});
|
||
}
|
||
|
||
private function markRefreshed(string $scope, array $parameters, array $metrics = []): void
|
||
{
|
||
$entryKey = $this->entryKey($scope, $parameters);
|
||
|
||
$this->updateRegistry(function (array $entries) use ($entryKey, $metrics): array {
|
||
if (! isset($entries[$entryKey])) {
|
||
return $entries;
|
||
}
|
||
|
||
$entries[$entryKey] = array_merge($entries[$entryKey], $metrics, ['urgent' => false]);
|
||
|
||
return $entries;
|
||
});
|
||
}
|
||
|
||
private function updateRegistry(Closure $callback): void
|
||
{
|
||
$lock = Cache::lock('jiansun:registry-lock:v2', 5);
|
||
|
||
try {
|
||
$lock->block(2, function () use ($callback): void {
|
||
$entries = Cache::get(self::REGISTRY_KEY, []);
|
||
$entries = is_array($entries) ? $entries : [];
|
||
$entries = $callback($entries);
|
||
|
||
uasort($entries, static fn (array $a, array $b): int => ((int) $b['seen_at']) <=> ((int) $a['seen_at']));
|
||
$limit = max((int) config('jiansun.snapshots.registry_limit', 300), 1);
|
||
|
||
Cache::forever(self::REGISTRY_KEY, array_slice($entries, 0, $limit, true));
|
||
});
|
||
} catch (LockTimeoutException) {
|
||
// 登记失败不能阻塞页面；下一次访问或刷新会再次尝试。
|
||
}
|
||
}
|
||
|
||
private function staleOrFail(string $staleKey, string $message, ?Throwable $previous = null): array
|
||
{
|
||
$stale = Cache::get($staleKey);
|
||
|
||
if ($this->hasValue($stale)) {
|
||
return ['value' => $stale['value'], 'status' => 'stale'];
|
||
}
|
||
|
||
throw new ServiceUnavailableHttpException(30, $message, $previous);
|
||
}
|
||
|
||
private function isCircuitOpen(): bool
|
||
{
|
||
return Cache::has(self::GLOBAL_CIRCUIT_KEY);
|
||
}
|
||
|
||
private function queryCooldownKey(string $queryId): string
|
||
{
|
||
return 'jiansun:query-cooldown:' . $queryId;
|
||
}
|
||
|
||
private function isScopeAllowed(string $scope): bool
|
||
{
|
||
$allowed = (array) config('jiansun.allowed_scopes', []);
|
||
|
||
foreach ($allowed as $pattern) {
|
||
$pattern = trim((string) $pattern);
|
||
|
||
if ($pattern === '*' || $pattern === $scope) {
|
||
return true;
|
||
}
|
||
|
||
if (str_ends_with($pattern, '.*') && str_starts_with($scope, substr($pattern, 0, -1))) {
|
||
return true;
|
||
}
|
||
}
|
||
|
||
return false;
|
||
}
|
||
|
||
private function entryKey(string $scope, array $parameters): string
|
||
{
|
||
return hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE));
|
||
}
|
||
|
||
private function cacheKey(string $scope, array $parameters): string
|
||
{
|
||
return 'jiansun:snapshot:' . self::CACHE_VERSION . ':' . $scope . ':'
|
||
. hash('sha256', json_encode($parameters, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION));
|
||
}
|
||
|
||
private function ttlFor(string $scope): int
|
||
{
|
||
if (str_ends_with($scope, '.options')) {
|
||
return (int) config('jiansun.snapshots.options_ttl_seconds', 3600);
|
||
}
|
||
|
||
if (str_ends_with($scope, '.summary')) {
|
||
return (int) config('jiansun.snapshots.summary_ttl_seconds', 120);
|
||
}
|
||
|
||
return (int) config('jiansun.snapshots.ttl_seconds', 120);
|
||
}
|
||
|
||
private function isDue(array $entry, int $now): bool
|
||
{
|
||
if ((bool) ($entry['urgent'] ?? false)) {
|
||
return true;
|
||
}
|
||
|
||
$scope = (string) $entry['scope'];
|
||
$parameters = (array) $entry['parameters'];
|
||
$cached = Cache::get($this->cacheKey($scope, $parameters));
|
||
|
||
if (! is_array($cached) || ! isset($cached['cached_at_ts'])) {
|
||
return true;
|
||
}
|
||
|
||
// 给一分钟任务留少量余量，使页面看到的数据通常保持在两分钟以内。
|
||
return ((int) $cached['cached_at_ts'] + max($this->ttlFor($scope), 60)) <= ($now + 15);
|
||
}
|
||
|
||
private function priorityFor(string $scope): int
|
||
{
|
||
return match (true) {
|
||
str_ends_with($scope, '.list') => 3,
|
||
str_ends_with($scope, '.summary') => 2,
|
||
default => 1,
|
||
};
|
||
}
|
||
|
||
private function rowCount(mixed $value): int
|
||
{
|
||
if (! is_array($value)) {
|
||
return 1;
|
||
}
|
||
|
||
if (isset($value['data']) && is_array($value['data'])) {
|
||
return count($value['data']);
|
||
}
|
||
|
||
return count($value);
|
||
}
|
||
|
||
private function hasValue(mixed $payload): bool
|
||
{
|
||
return is_array($payload) && array_key_exists('value', $payload);
|
||
}
|
||
|
||
private function setReadState(string $status, array $payload): void
|
||
{
|
||
$this->lastReadState = [
|
||
'status' => $status,
|
||
'cached_at' => $payload['cached_at'] ?? null,
|
||
];
|
||
}
|
||
|
||
private function normalize(array $parameters): array
|
||
{
|
||
foreach ($parameters as &$value) {
|
||
if (is_array($value)) {
|
||
$value = $this->normalize($value);
|
||
}
|
||
}
|
||
unset($value);
|
||
|
||
ksort($parameters);
|
||
|
||
return $parameters;
|
||
}
|
||
}
|