优化:异步刷新上游财务数据快照

This commit is contained in:
WuYongJie 2026-07-22 10:25:52 +08:00
parent 49f9770785
commit 809d6c5975
12 changed files with 563 additions and 139 deletions

View File

@ -123,12 +123,15 @@
### 8.6 上游数据库读取规则
- 业务页面禁止直接、无缓存地查询上游数据库;统一通过 `JianSunQueryCache` 读取本地活跃查询快照。
- 相同页面和筛选条件默认复用 15 分钟,选项类数据默认复用 60 分钟;不得通过前端轮询缩短该周期。
- 所有上游查询全局单线程执行,必须设置连接超时、查询超时和每 15 分钟刷新预算;失败时只允许使用已有旧快照,不得无限重试。
- 定时任务只刷新最近 24 小时内实际使用过的查询,默认每轮最多 12 个;闲置页面不得为了“预热”持续访问上游。
- `JIANSUN_ENABLED` 默认关闭。只有确认生产服务器网络白名单、只读账号、ODBC 驱动和查询索引后,才能在服务器私有 `runtime/.env` 中开启。
- 真正的行级增量同步必须先确认每张源表的稳定主键、更新时间字段、删除标记和对应索引;未确认前禁止用全表扫描冒充增量同步。
- 上游是长期运行且不可修改的旧系统,按只读黑盒处理;不得要求修改其账号、表结构、索引、字段或启用 Change Tracking 等功能。
- 业务页面禁止直接、无缓存地查询上游数据库;统一通过 `JianSunQueryCache` 读取持久化在本机 `storage` 卷中的活跃查询快照。
- 快照复用按“数据范围 + 查询条件”判断,与账户数量无关;权限或数据范围不同的结果不得错误共用。
- 列表和汇总快照默认 2 分钟,选项类数据默认 60 分钟。页面命中过期快照时立即返回本地值并排入后台刷新,不得等待上游,也不得通过前端高频轮询制造压力。
- 相同查询使用独立单飞锁;不同冷查询最多占用 2 个上游槽位。每分钟最多 6 个查询组,每分钟后台最多刷新 2 个近期活跃查询;不得恢复为一把锁串行所有页面请求。
- 必须设置连接超时、查询超时、慢查询冷却和连接失败熔断;一次失败后进入冷却,已有旧快照时继续提供本地数据,不得连续重试。
- 上游 API 必须保留 `X-Caiwu-Data-Status` 和 `X-Caiwu-Data-Synced-At` 响应头;验收同时检查数据状态和同步时间,不能只看 HTTP 200。
- `JIANSUN_ENABLED` 默认关闭;只允许在现有只读访问方式完成受控连通性和单查询探测后,通过服务器私有 `runtime/.env` 开启。
- 当前方案是近实时查询快照,不得冒充完整行级增量复制。没有可靠变更标记时,禁止定时全表扫描;历史一致性只能采用低频、小批量、可中止的核对。
- 首次全量历史数据同步必须单独评估行数、时间窗口和上游低峰期,不得在日常发布时自动执行。
- 上游地址、账号、密码只允许存在于服务器私有环境配置;代码、提交、PR、共享文档和日志中一律不得出现真实值。

View File

@ -13,7 +13,7 @@ class JianSunRefreshActiveSnapshots extends Command
{
protected $signature = 'jiansun:snapshots:refresh {--limit= : 本轮最多刷新的活跃查询数}';
protected $description = '单线程刷新最近使用过的 JianSun 上游查询快照';
protected $description = '在时间和并发预算内刷新最近使用过的 JianSun 本地快照';
public function handle(JianSunQueryCache $cache, JianSunSnapshotRefresher $refresher): int
{
@ -23,9 +23,11 @@ class JianSunRefreshActiveSnapshots extends Command
return self::SUCCESS;
}
$configuredLimit = max((int) config('jiansun.snapshots.refresh_limit', 12), 1);
$configuredLimit = max((int) config('jiansun.snapshots.refresh_limit', 2), 1);
$requestedLimit = (int) ($this->option('limit') ?: $configuredLimit);
$entries = $cache->activeEntries(min(max($requestedLimit, 1), $configuredLimit));
$timeBudget = max((int) config('jiansun.snapshots.refresh_time_budget_seconds', 45), 10);
$startedAt = microtime(true);
if ($entries === []) {
$this->components->info('没有需要刷新的活跃查询。');
@ -37,6 +39,11 @@ class JianSunRefreshActiveSnapshots extends Command
$stale = 0;
foreach ($entries as $entry) {
if ((microtime(true) - $startedAt) >= $timeBudget) {
$this->components->info('本轮刷新已达到时间预算,剩余查询留到下一轮。');
break;
}
try {
$status = $refresher->refresh($entry);
$status === 'fresh' ? $fresh++ : $stale++;

View File

@ -29,22 +29,31 @@ return [
| 活跃查询快照
|--------------------------------------------------------------------------
|
| 页面只在快照过期时访问一次上游;相同页面和筛选条件由所有用户共享。
| 全局锁保证同一时间只有一个查询组访问上游,预算则限制十五分钟内的
| 最大刷新次数。旧快照仅在上游临时不可用时作为明确的降级数据使用。
| 页面优先读取持久化在本机 storage 卷中的快照;快照过期后先返回旧值,
| 再由后台刷新。相同数据范围和筛选条件复用同一快照,与账户数量无关。
| 同一查询只允许一个刷新任务,不同查询最多占用少量上游槽位,并受
| 每分钟预算、慢查询冷却和连接失败熔断共同保护。
|
*/
'snapshots' => [
'ttl_seconds' => (int) env('JIANSUN_SNAPSHOT_TTL', 900),
'ttl_seconds' => (int) env('JIANSUN_SNAPSHOT_TTL', 120),
'summary_ttl_seconds' => (int) env('JIANSUN_SUMMARY_TTL', 120),
'options_ttl_seconds' => (int) env('JIANSUN_OPTIONS_TTL', 3600),
'stale_ttl_seconds' => (int) env('JIANSUN_STALE_TTL', 86400),
'active_ttl_seconds' => (int) env('JIANSUN_ACTIVE_QUERY_TTL', 86400),
'registry_limit' => (int) env('JIANSUN_ACTIVE_QUERY_LIMIT', 60),
'refresh_limit' => (int) env('JIANSUN_REFRESH_LIMIT', 12),
'miss_budget_per_window' => (int) env('JIANSUN_MISS_BUDGET', 20),
'budget_window_seconds' => (int) env('JIANSUN_BUDGET_WINDOW', 900),
'active_ttl_seconds' => (int) env('JIANSUN_ACTIVE_QUERY_TTL', 21600),
'registry_limit' => (int) env('JIANSUN_ACTIVE_QUERY_LIMIT', 300),
'refresh_limit' => (int) env('JIANSUN_REFRESH_LIMIT', 2),
'miss_budget_per_window' => (int) env('JIANSUN_MISS_BUDGET', 6),
'budget_window_seconds' => (int) env('JIANSUN_BUDGET_WINDOW', 60),
'lock_seconds' => (int) env('JIANSUN_LOCK_SECONDS', 45),
'lock_wait_seconds' => (int) env('JIANSUN_LOCK_WAIT', 12),
'lock_wait_seconds' => (int) env('JIANSUN_LOCK_WAIT', 3),
'max_concurrency' => (int) env('JIANSUN_MAX_CONCURRENCY', 2),
'slot_wait_ms' => (int) env('JIANSUN_SLOT_WAIT_MS', 1500),
'failure_cooldown_seconds' => (int) env('JIANSUN_FAILURE_COOLDOWN', 300),
'query_failure_cooldown_seconds' => (int) env('JIANSUN_QUERY_FAILURE_COOLDOWN', 300),
'slow_query_ms' => (int) env('JIANSUN_SLOW_QUERY_MS', 5000),
'slow_query_cooldown_seconds' => (int) env('JIANSUN_SLOW_QUERY_COOLDOWN', 300),
'refresh_time_budget_seconds' => (int) env('JIANSUN_REFRESH_TIME_BUDGET', 45),
'refresh_pause_ms' => (int) env('JIANSUN_REFRESH_PAUSE_MS', 300),
],
];

View File

@ -78,7 +78,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha"
# 只运行 JianSun 快照刷新命令,不启用整个 Laravel 调度器,避免意外启动
# 账单提醒等既有任务。命令在 JIANSUN_ENABLED=false 时会安全跳过。
cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF'
*/15 * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1
* * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1
EOF
chmod 644 /etc/cron.d/caiwu-jiansun-snapshots

View File

@ -82,7 +82,7 @@ mv "$REMOTE_DIR/.deployed-sha.tmp" "$REMOTE_DIR/.deployed-sha"
# 只运行 JianSun 快照刷新命令,不启用整个 Laravel 调度器,避免意外启动
# 账单提醒等既有任务。命令在 JIANSUN_ENABLED=false 时会安全跳过。
cat > /etc/cron.d/caiwu-jiansun-snapshots <<'EOF'
*/15 * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1
* * * * * root /usr/bin/flock -n /run/caiwu-jiansun-sync.lock /usr/bin/docker exec -u www-data caiwu_app php artisan jiansun:snapshots:refresh >/dev/null 2>&1
EOF
chmod 644 /etc/cron.d/caiwu-jiansun-snapshots

View File

@ -58,6 +58,7 @@ if [ ! -f "$REMOTE_DIR/runtime/.env" ]; then
'CACHE_DRIVER=file' \
'SESSION_DRIVER=file' \
'QUEUE_CONNECTION=sync' \
'JIANSUN_ENABLED=false' \
'MARIADB_DATABASE=caiwu' \
'MARIADB_USER=caiwu' \
"MARIADB_PASSWORD=$DB_PASSWORD" \

View File

@ -137,17 +137,20 @@ git pull --ff-only
## 8. 上游财务数据库同步
生产环境默认设置 `JIANSUN_ENABLED=false`。网络白名单、只读账号和索引未核实前,即使镜像已经包含 ODBC 驱动,也不会连接上游。
生产环境默认设置 `JIANSUN_ENABLED=false`。上游按不可修改的只读旧系统处理,不要求对方修改账号、表结构、索引、字段或数据库功能;只有现有访问方式通过受控连通性和单查询探测后,才在服务器私有配置中开启。
启用后采用“活跃查询快照”策略:
1. 页面相同查询 15 分钟内直接读取服务器本地文件缓存,6 个测试账号共享同一份结果;
2. 下拉选项等低频数据缓存 60 分钟;
3. `/etc/cron.d/caiwu-jiansun-snapshots` 每 15 分钟触发一次刷新,只处理最近使用过的查询;
4. 所有刷新使用同一把互斥锁,上游同时最多只有 1 个查询组;
5. 每轮默认最多刷新 12 个活跃查询,每 15 分钟默认最多产生 20 次实际缓存未命中;
6. 连接 5 秒未建立或单次查询超过 15 秒即失败,不连续重试;已有旧快照时明确降级为旧数据;
7. 服务器和上游不通时保持关闭,页面不会绕过缓存直接连接。
1. 页面优先读取持久化在 `caiwu_storage` 卷中的本地快照;相同数据范围和查询条件复用同一结果,与系统账户数量无关;
2. 列表和汇总默认保持 2 分钟新鲜,下拉选项等低频数据缓存 60 分钟;
3. 新鲜快照直接返回;过期快照也立即返回,同时登记为紧急后台刷新,页面不等待上游;只有本地从未存在该查询时,才允许受控同步获取一次;
4. `/etc/cron.d/caiwu-jiansun-snapshots` 每分钟触发一次,只处理最近 6 小时实际使用且已经到期的查询;
5. 相同查询使用独立单飞锁,不同查询最多占用 2 个上游槽位,因此大量缓存命中不会被全局串行;
6. 每轮默认最多刷新 2 个活跃查询,每分钟默认最多产生 6 个实际上游查询组;列表优先于汇总,汇总优先于选项;
7. 连接 5 秒未建立或单次查询超过 15 秒即失败;慢查询和失败查询进入 5 分钟冷却,连接失败同时触发全局熔断,不连续重试;
8. 服务器和上游不通时继续使用已有本地快照;完全没有本地数据时明确返回暂不可用,不绕过保护直接连接。
相关 API 使用 `X-Caiwu-Data-Status`(`fresh` / `stale`)和 `X-Caiwu-Data-Synced-At` 响应头报告数据状态,验收时不能只看 HTTP 200。
服务器私有 `runtime/.env` 只使用下列变量名,真实值不得写入 Git 或本文档:
@ -159,13 +162,22 @@ DB_JIANSUN_DATABASE=
DB_JIANSUN_USERNAME=
DB_JIANSUN_PASSWORD=
DB_JIANSUN_ODBC_DRIVER=FreeTDS
JIANSUN_SNAPSHOT_TTL=900
JIANSUN_SNAPSHOT_TTL=120
JIANSUN_SUMMARY_TTL=120
JIANSUN_OPTIONS_TTL=3600
JIANSUN_REFRESH_LIMIT=12
JIANSUN_MISS_BUDGET=20
JIANSUN_ACTIVE_QUERY_TTL=21600
JIANSUN_ACTIVE_QUERY_LIMIT=300
JIANSUN_REFRESH_LIMIT=2
JIANSUN_MISS_BUDGET=6
JIANSUN_BUDGET_WINDOW=60
JIANSUN_MAX_CONCURRENCY=2
JIANSUN_FAILURE_COOLDOWN=300
JIANSUN_QUERY_FAILURE_COOLDOWN=300
JIANSUN_SLOW_QUERY_MS=5000
JIANSUN_SLOW_QUERY_COOLDOWN=300
```
这套策略控制的是当前页面查询压力,不等同于复制上游全库。若以后需要真正的行级增量镜像,必须先由负责人确认每张表的稳定主键、更新时间、删除标记和索引,再安排一次低峰期初始化;禁止对数百万行大表每 10~15 分钟做全表扫描。
这套策略提供的是近实时查询快照,不等同于复制上游全库。对于没有可靠更新时间或删除标记的旧表,不能声称实现了完整行级增量;禁止为了追求秒级实时而周期性扫描数百万行。页面实时性依靠本地快照和一分钟后台刷新,历史一致性只能通过低频、小批量、可随时停止的核对补齐。
## 9. 禁止事项

View File

@ -41,7 +41,7 @@ class GoodsReconciliations extends ApiController
fn () => $this->repo->list($filters, $page, $limit)
);
return response()->json([
return $this->snapshotResponse([
'data' => $result['data'],
'total' => $result['total'],
'page' => $page,
@ -62,7 +62,7 @@ class GoodsReconciliations extends ApiController
fn () => $this->repo->summary($filters)
);
return response()->json($summary);
return $this->snapshotResponse($summary);
}
/**
@ -76,13 +76,24 @@ class GoodsReconciliations extends ApiController
fn () => $this->repo->filterOptions()
);
return response()->json($options);
return $this->snapshotResponse($options);
}
// ===================================================================
// Helpers
// ===================================================================
private function snapshotResponse(array $payload)
{
$response = response()->json($payload);
foreach ($this->cache->responseHeaders() as $name => $value) {
$response->headers->set($name, $value);
}
return $response;
}
/**
* 从请求参数构建筛选条件
*/

View File

@ -339,7 +339,7 @@ class SupplierPayables extends ApiController
fn () => $resolver($page, $limit)
);
return response()->json([
return $this->snapshotResponse([
'data' => $result['data'],
'total' => $result['total'],
'page' => $page,
@ -350,7 +350,7 @@ class SupplierPayables extends ApiController
private function summaryResponse(string $scope, array $filters, Closure $resolver)
{
return response()->json($this->cache->remember(
return $this->snapshotResponse($this->cache->remember(
$scope . '.summary',
['filters' => $filters],
$resolver
@ -359,13 +359,24 @@ class SupplierPayables extends ApiController
private function optionsResponse(string $scope, Closure $resolver)
{
return response()->json($this->cache->remember(
return $this->snapshotResponse($this->cache->remember(
$scope . '.options',
[],
$resolver
));
}
private function snapshotResponse(array $payload)
{
$response = response()->json($payload);
foreach ($this->cache->responseHeaders() as $name => $value) {
$response->headers->set($name, $value);
}
return $response;
}
private function filters(string $type): array
{
$filters = [];

View File

@ -29,7 +29,14 @@ class SupplierPayableRepository
/**
* 通用工具:带筛选 + 分页的查询
*/
private function runQuery(string $sql, array $params, int $page, int $pageSize, int $countThreshold = 5000): array
private function runQuery(
string $sql,
array $params,
int $page,
int $pageSize,
string $orderBy = '(SELECT 0)',
int $countThreshold = 5000
): array
{
$offset = ($page - 1) * $pageSize;
@ -45,8 +52,9 @@ class SupplierPayableRepository
$total = -1;
}
// 2. 分页数据(用 TOP + OFFSET/FETCH 不可行,统一使用 ROW_NUMBER() OVER 包装)
$pagedSql = 'SELECT * FROM (SELECT *, ROW_NUMBER() OVER (ORDER BY (SELECT 0)) AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ?';
// 2. 分页数据:排序字段只允许由仓库内部固定传入,不接受请求参数。
// ROW_NUMBER 位于无 ORDER BY 的派生表外层,兼容旧版 SQL Server。
$pagedSql = 'SELECT * FROM (SELECT src.*, ROW_NUMBER() OVER (ORDER BY ' . $orderBy . ') AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ? ORDER BY rn';
$paramsWithRange = array_merge($params, [$offset + 1, $offset + $pageSize]);
$stmt = $this->db()->prepare($pagedSql);
@ -61,10 +69,17 @@ class SupplierPayableRepository
/**
* 计算分页的 SELECT TOP N(无 COUNT),加速大表查询
*/
private function runQueryFast(string $sql, array $params, int $page, int $pageSize, int $total = -1): array
private function runQueryFast(
string $sql,
array $params,
int $page,
int $pageSize,
int $total = -1,
string $orderBy = '(SELECT 0)'
): array
{
$offset = ($page - 1) * $pageSize;
$pagedSql = 'SELECT * FROM (SELECT *, ROW_NUMBER() OVER (ORDER BY (SELECT 0)) AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ?';
$pagedSql = 'SELECT * FROM (SELECT src.*, ROW_NUMBER() OVER (ORDER BY ' . $orderBy . ') AS rn FROM (' . $sql . ') AS src) AS t WHERE rn BETWEEN ? AND ? ORDER BY rn';
$paramsWithRange = array_merge($params, [$offset + 1, $offset + $pageSize]);
$stmt = $this->db()->prepare($pagedSql);
$stmt->execute($paramsWithRange);
@ -135,7 +150,7 @@ class SupplierPayableRepository
FROM finance_purchaseReturn
$whereSql";
$result = $this->runQuery($sql, $params, $page, $pageSize);
$result = $this->runQuery($sql, $params, $page, $pageSize, 'send_time DESC, wms_pick_no DESC');
return [
'data' => $result['data'],
@ -217,7 +232,7 @@ class SupplierPayableRepository
FROM finance_invoiceDeduction
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, create_time DESC');
}
public function invoiceDeductionSummary(array $filters): array
@ -292,10 +307,9 @@ class SupplierPayableRepository
create_time
FROM fys_pressing_factory_compensation_detail
) AS t
$whereSql
ORDER BY month DESC, create_time DESC";
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'month DESC, create_time DESC');
}
public function compensationDetailSummary(array $filters): array
@ -374,10 +388,9 @@ class SupplierPayableRepository
CONVERT(varchar(19), createTime, 120) AS create_time,
creator
FROM finance_advanceLoans
$whereSql
ORDER BY month DESC, createTime DESC";
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, create_time DESC');
}
public function advanceDetailSummary(array $filters): array
@ -455,10 +468,9 @@ class SupplierPayableRepository
businessId,
CONVERT(varchar(19), updateAt, 120) AS update_time
FROM finance_sampleDeduction
$whereSql
ORDER BY finishTime DESC";
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'finish_time DESC');
}
public function sampleDeductionSummary(array $filters): array
@ -542,10 +554,9 @@ class SupplierPayableRepository
CONVERT(varchar(10), shipDate, 120) AS ship_date,
expressName
FROM pm_unOrderDetails
$whereSql
ORDER BY date DESC, Id DESC";
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'date DESC, id DESC');
}
public function deliveryUnpricedSummary(array $filters): array
@ -622,10 +633,9 @@ class SupplierPayableRepository
labels,
creator_name AS creator
FROM finance_purchaseReturn
$whereSql
ORDER BY date DESC";
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'send_time DESC');
}
public function returnUnpricedSummary(array $filters): array
@ -753,10 +763,9 @@ class SupplierPayableRepository
mark AS remark2,
checker AS person_charge
FROM fys_supplierPaymentSummary
$whereSql
ORDER BY Month DESC, supplier";
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, supplier');
}
public function otherAdjustmentsSummary(array $filters): array
@ -830,10 +839,9 @@ class SupplierPayableRepository
creator,
CONVERT(varchar(19), create_time, 120) AS create_time
FROM fys_express_courier_settlement
$whereSql
ORDER BY month DESC, expressname";
$whereSql";
return $this->runQuery($sql, $params, $page, $pageSize);
return $this->runQuery($sql, $params, $page, $pageSize, 'month_str DESC, express_name');
}
public function dropshippingExpressSummary(array $filters): array

View File

@ -3,6 +3,7 @@
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;
@ -10,19 +11,23 @@ use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
use Throwable;
/**
* 上游查询的本地快照与单飞保护。
* 上游查询的本地快照、请求合并和压力保护。
*
* 这不是上游全库镜像:它只保留最近实际使用过的页面查询,避免 2GB
* 服务器复制数百万行,同时确保多个用户不会重复冲击上游数据库。
* 页面命中新鲜快照时立即返回;命中过期快照时也立即返回,并把刷新登记给
* 后台任务。只有完全没有本地快照的冷查询才会在请求内受控访问一次上游。
*/
class JianSunQueryCache
{
private const CACHE_VERSION = 'v1';
private const CACHE_VERSION = 'v2';
private const REGISTRY_KEY = 'jiansun:snapshots:active';
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
{
@ -30,20 +35,50 @@ class JianSunQueryCache
$key = $this->cacheKey($scope, $parameters);
$cached = Cache::get($key);
if (is_array($cached) && array_key_exists('value', $cached)) {
$this->register($scope, $parameters);
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'];
}
// 首次访问没有任何本地数据时才允许同步获取一次;先登记,失败后后台
// 仍可在熔断结束后继续补齐快照。
$this->register($scope, $parameters, true);
$resolved = $this->resolve($scope, $parameters, $resolver, false);
$this->register($scope, $parameters);
$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
*/
@ -55,12 +90,12 @@ class JianSunQueryCache
}
/**
* 返回最近实际访问过的查询,供十五分钟定时任务逐个刷新。
* 返回最近实际访问且已经到期的查询,紧急项和列表优先。
*/
public function activeEntries(?int $limit = null): array
{
$now = time();
$activeTtl = max((int) config('jiansun.snapshots.active_ttl_seconds', 86400), 900);
$activeTtl = max((int) config('jiansun.snapshots.active_ttl_seconds', 21600), 900);
$entries = Cache::get(self::REGISTRY_KEY, []);
if (! is_array($entries)) {
@ -74,9 +109,23 @@ class JianSunQueryCache
&& $this->isDue($entry, $now);
});
usort($entries, static fn (array $a, array $b): int => $b['seen_at'] <=> $a['seen_at']);
usort($entries, function (array $a, array $b): int {
$urgent = ((int) ($b['urgent'] ?? 0)) <=> ((int) ($a['urgent'] ?? 0));
$limit ??= max((int) config('jiansun.snapshots.refresh_limit', 12), 1);
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);
}
@ -85,38 +134,53 @@ class JianSunQueryCache
{
$key = $this->cacheKey($scope, $parameters);
$staleKey = $key . ':stale';
$lock = Cache::lock(
'jiansun:upstream-query-lock',
$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 $lock->block(
max((int) config('jiansun.snapshots.lock_wait_seconds', 12), 1),
function () use ($scope, $parameters, $resolver, $force, $key, $staleKey): array {
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 (is_array($cached) && array_key_exists('value', $cached)) {
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) {
Log::warning('JianSun 上游查询失败,尝试读取旧快照。', [
'scope' => $scope,
'exception' => $exception::class,
]);
$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(),
@ -130,11 +194,168 @@ class JianSunQueryCache
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);
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) {
// 登记失败不能阻塞页面;下一次访问或刷新会再次尝试。
}
}
@ -142,62 +363,26 @@ class JianSunQueryCache
{
$stale = Cache::get($staleKey);
if (is_array($stale) && array_key_exists('value', $stale)) {
if ($this->hasValue($stale)) {
return ['value' => $stale['value'], 'status' => 'stale'];
}
throw new ServiceUnavailableHttpException(30, $message, $previous);
}
/**
* 所有刷新都在同一把全局锁内执行,因此这里的计数读写不会并发冲突。
*/
private function consumeBudget(): bool
private function isCircuitOpen(): bool
{
$window = max((int) config('jiansun.snapshots.budget_window_seconds', 900), 60);
$limit = max((int) config('jiansun.snapshots.miss_budget_per_window', 20), 1);
$bucket = intdiv(time(), $window);
$key = 'jiansun:upstream-budget:' . $bucket;
$count = (int) Cache::get($key, 0);
if ($count >= $limit) {
return false;
}
Cache::put($key, $count + 1, $window + 60);
return true;
return Cache::has(self::GLOBAL_CIRCUIT_KEY);
}
private function register(string $scope, array $parameters): void
private function queryCooldownKey(string $queryId): string
{
$entryKey = hash('sha256', $scope . '|' . json_encode($parameters, JSON_UNESCAPED_UNICODE));
return 'jiansun:query-cooldown:' . $queryId;
}
// 同一查询五分钟内只更新一次活跃时间,减少文件缓存写入。
if (! Cache::add('jiansun:registry-touch:' . $entryKey, true, 300)) {
return;
}
$lock = Cache::lock('jiansun:registry-lock', 5);
try {
$lock->block(2, function () use ($entryKey, $scope, $parameters): void {
$entries = Cache::get(self::REGISTRY_KEY, []);
$entries = is_array($entries) ? $entries : [];
$entries[$entryKey] = [
'scope' => $scope,
'parameters' => $parameters,
'seen_at' => time(),
];
uasort($entries, static fn (array $a, array $b): int => $b['seen_at'] <=> $a['seen_at']);
$limit = max((int) config('jiansun.snapshots.registry_limit', 60), 1);
Cache::forever(self::REGISTRY_KEY, array_slice($entries, 0, $limit, true));
});
} catch (LockTimeoutException) {
// 注册失败不应阻塞页面;下一次请求会再次尝试。
}
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
@ -212,11 +397,19 @@ class JianSunQueryCache
return (int) config('jiansun.snapshots.options_ttl_seconds', 3600);
}
return (int) config('jiansun.snapshots.ttl_seconds', 900);
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));
@ -225,8 +418,43 @@ class JianSunQueryCache
return true;
}
// 给 cron 一分钟余量,避免刚好在过期边界前运行后再等十五分钟。
return ((int) $cached['cached_at_ts'] + max($this->ttlFor($scope), 60)) <= ($now + 60);
// 给一分钟任务留少量余量,使页面看到的数据通常保持在两分钟以内。
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

View File

@ -0,0 +1,134 @@
<?php
namespace Tests\Unit\EcommerceFinance;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\Cache;
use Modules\EcommerceFinance\Services\JianSunQueryCache;
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
use Tests\TestCase;
class JianSunQueryCacheTest extends TestCase
{
protected function setUp(): void
{
parent::setUp();
Cache::setDefaultDriver('array');
Cache::flush();
Carbon::setTestNow('2026-07-22 10:00:00');
config()->set('jiansun.snapshots.ttl_seconds', 60);
config()->set('jiansun.snapshots.summary_ttl_seconds', 60);
config()->set('jiansun.snapshots.options_ttl_seconds', 3600);
config()->set('jiansun.snapshots.stale_ttl_seconds', 3600);
config()->set('jiansun.snapshots.active_ttl_seconds', 3600);
config()->set('jiansun.snapshots.registry_limit', 300);
config()->set('jiansun.snapshots.refresh_limit', 2);
config()->set('jiansun.snapshots.miss_budget_per_window', 1000);
config()->set('jiansun.snapshots.budget_window_seconds', 60);
config()->set('jiansun.snapshots.max_concurrency', 2);
config()->set('jiansun.snapshots.slot_wait_ms', 0);
config()->set('jiansun.snapshots.lock_wait_seconds', 1);
config()->set('jiansun.snapshots.slow_query_ms', 5000);
}
protected function tearDown(): void
{
Carbon::setTestNow();
Cache::flush();
parent::tearDown();
}
public function test_identical_queries_are_shared_without_user_count_limit(): void
{
$cache = app(JianSunQueryCache::class);
$calls = 0;
for ($i = 0; $i < 100; $i++) {
$value = $cache->remember('goods.list', ['page' => 1], function () use (&$calls): array {
$calls++;
return ['data' => [['id' => 1]], 'total' => 1];
});
$this->assertSame(1, $value['total']);
}
$this->assertSame(1, $calls);
}
public function test_expired_snapshot_returns_immediately_and_is_refreshed_later(): void
{
$cache = app(JianSunQueryCache::class);
$calls = 0;
$first = $cache->remember('goods.list', ['page' => 1], function () use (&$calls): array {
$calls++;
return ['data' => [['version' => 1]], 'total' => 1];
});
Carbon::setTestNow(now()->addSeconds(61));
$stale = $cache->remember('goods.list', ['page' => 1], function () use (&$calls): array {
$calls++;
return ['data' => [['version' => 999]], 'total' => 1];
});
$this->assertSame($first, $stale);
$this->assertSame(1, $calls, '页面读取过期快照时不应同步访问上游。');
$this->assertSame('stale', $cache->responseHeaders()['X-Caiwu-Data-Status']);
$this->assertTrue((bool) ($cache->activeEntries(1)[0]['urgent'] ?? false));
$status = $cache->refresh('goods.list', ['page' => 1], function () use (&$calls): array {
$calls++;
return ['data' => [['version' => 2]], 'total' => 1];
});
$fresh = $cache->remember('goods.list', ['page' => 1], fn (): array => []);
$this->assertSame('fresh', $status);
$this->assertSame(2, $fresh['data'][0]['version']);
$this->assertSame('fresh', $cache->responseHeaders()['X-Caiwu-Data-Status']);
$this->assertSame(2, $calls);
}
public function test_unrelated_cold_queries_do_not_share_one_global_query_lock(): void
{
$cache = app(JianSunQueryCache::class);
$parametersA = ['page' => 1];
$queryA = hash('sha256', 'goods.list|' . json_encode($parametersA, JSON_UNESCAPED_UNICODE));
$lockA = Cache::lock('jiansun:query-lock:' . $queryA, 30);
$this->assertTrue($lockA->get());
try {
$value = $cache->remember('goods.summary', [], fn (): array => ['total' => 7]);
$this->assertSame(7, $value['total']);
} finally {
$lockA->release();
}
}
public function test_upstream_slots_bound_parallel_cold_queries(): void
{
$cache = app(JianSunQueryCache::class);
$slot0 = Cache::lock('jiansun:upstream-slot:0', 30);
$slot1 = Cache::lock('jiansun:upstream-slot:1', 30);
$this->assertTrue($slot0->get());
$this->assertTrue($slot1->get());
try {
$this->expectException(ServiceUnavailableHttpException::class);
$cache->remember('goods.list', ['page' => 2], fn (): array => ['data' => [], 'total' => 0]);
} finally {
$slot0->release();
$slot1->release();
}
}
}