From 69e3d9a396e2251260a0cda9f452fdff449fb8c8 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sat, 3 Oct 2026 15:23:11 +0800 Subject: [PATCH] =?UTF-8?q?perf(worker):=20=E3=80=8C=E6=88=91=E7=9A=84?= =?UTF-8?q?=E8=AE=A2=E5=8D=95=E3=80=8D=E7=BC=93=E5=AD=98=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E9=A9=B1=E5=8A=A8=E5=A4=B1=E6=95=88=EF=BC=8C?= =?UTF-8?q?TTL=202s/5s=20=E2=86=92=2030s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 这是当前最大瓶颈的根治：打手端三件套（列表/计数/概览）占全库约 75% 缓冲读， 单查询成本是 O(打手单量) 的固有成本（重打手 8,586 单逐行探测）， 索引层面已无可抠，只能把"计算次数"和"访问次数"解耦。 原先 TTL 只有 2s/5s，是因为文档担心"再长会让分页总数滞后"。 但该数据只在该打手自己的工单流转时才变，而项目已有 realtime 快照广播 精确感知每次流转——把同一事件流接到缓存失效上，长 TTL 与新鲜度可以兼得， 正确性反而优于固定 5s（那同样存在滞后窗口）。 - WorkerListCache 新增 evictByPrefix：按键前缀整体逐出； - 缓存实例抽到中立模块 worker-my-order-cache.ts：realtime 需要引用失效入口， 而 my-order-service 已 import realtime，实例放任何一方都会循环依赖； - 缓存键统一为 ["w",workerId,scope,detail]，前缀失效不会误伤其它打手； - 失效挂在 publishWorkOrderSnapshots：那里恰好已有工单 + 全部拼单人， 受影响集合取 当前指派 + 历史保留 + 全部拼单人 + 调用方指定， 21 处工单写入路径（含取消申请/反馈处理）都会经过； 工单已删除的分支同样失效，避免打手继续看到已删的单。 最坏情况（漏失效）是打手最多 30s 看到旧状态，与原 5s TTL 的滞后窗口同性质。 守卫测试锁定：失效入口存在、键前缀格式、受影响集合三要素齐全。 --- .../realtime/realtime-event-service.ts | 16 +++++ .../worker-platform/worker-list-cache.ts | 16 +++++ .../worker-my-order-cache.test.ts | 62 +++++++++++++++++++ .../worker-platform/worker-my-order-cache.ts | 50 +++++++++++++++ .../worker-my-order-service.ts | 30 ++++----- 5 files changed, 157 insertions(+), 17 deletions(-) create mode 100644 apps/backend/src/services/worker-platform/worker-my-order-cache.test.ts create mode 100644 apps/backend/src/services/worker-platform/worker-my-order-cache.ts diff --git a/apps/backend/src/services/realtime/realtime-event-service.ts b/apps/backend/src/services/realtime/realtime-event-service.ts index 4a36a7a1..a149dd8d 100644 --- a/apps/backend/src/services/realtime/realtime-event-service.ts +++ b/apps/backend/src/services/realtime/realtime-event-service.ts @@ -29,6 +29,7 @@ import { mapWorkerUser, resolveWorkerPermissions, } from '../worker-platform/mappers.js' +import { invalidateWorkerMyOrderCaches } from '../worker-platform/worker-my-order-cache.js' export type RealtimeEventType = | 'admin_notification.changed' @@ -249,6 +250,9 @@ async function publishWorkOrderSnapshots(input: { const workOrderId = Math.max(0, Number(input.workOrderId) || 0) const workOrder = await getWorkOrderById(workOrderId) if (!workOrder) { + // 工单已不存在（删除）：调用方传入的打手集合仍需失效缓存， + // 否则他们的「我的订单」会继续展示已删除的单。 + invalidateWorkerMyOrderCaches(input.workerIds || []) publishRealtimeEvent({ type: 'work_order.changed', entityId: workOrderId, @@ -262,6 +266,18 @@ async function publishWorkOrderSnapshots(input: { const pendingUnfreezeByOrderId = await sumPendingUnfreezeByOrderIds([workOrderId]) const shares = await listWorkOrderShares(workOrderId) + + // 事件驱动缓存失效：这里已经拿到的工单与拼单快照恰好给出完整的受影响打手集合 + // （当前指派 + 历史保留 + 全部拼单人 + 调用方显式指定的）。 + // 先失效再广播，保证打手端随后的拉取一定看到新数据；这也是「我的订单」缓存 + // 能放心用 30s 长 TTL 的前提（见 worker-my-order-cache.ts）。 + invalidateWorkerMyOrderCaches([ + ...(input.workerIds || []), + workOrder.assigned_worker_id, + workOrder.last_assigned_worker_id, + ...shares.map((share) => share.worker_id), + ]) + const joinedQuantity = shares .filter((share) => share.status !== 'cancelled') .reduce((sum, share) => sum + Number(share.quantity || 0), 0) diff --git a/apps/backend/src/services/worker-platform/worker-list-cache.ts b/apps/backend/src/services/worker-platform/worker-list-cache.ts index b9f9521a..2bb61583 100644 --- a/apps/backend/src/services/worker-platform/worker-list-cache.ts +++ b/apps/backend/src/services/worker-platform/worker-list-cache.ts @@ -29,4 +29,20 @@ export class WorkerListCache { this.entries.set(key, { expiresAt: now + this.ttlMs, value }) return value } + + /** + * 逐出所有键以指定前缀开头的缓存项，返回逐出数量。 + * 供事件驱动失效使用：数据变更时按主体（如打手 id）整体作废， + * 让长 TTL 与新鲜度可以兼得。 + */ + evictByPrefix(prefix: string): number { + let evicted = 0 + for (const key of this.entries.keys()) { + if (key.startsWith(prefix)) { + this.entries.delete(key) + evicted += 1 + } + } + return evicted + } } diff --git a/apps/backend/src/services/worker-platform/worker-my-order-cache.test.ts b/apps/backend/src/services/worker-platform/worker-my-order-cache.test.ts new file mode 100644 index 00000000..3ffb6340 --- /dev/null +++ b/apps/backend/src/services/worker-platform/worker-my-order-cache.test.ts @@ -0,0 +1,62 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { WorkerListCache } from './worker-list-cache.js' + +test('evictByPrefix 只逐出匹配前缀的条目，不影响其它打手', async () => { + const cache = new WorkerListCache(60_000) + let workerA = 0 + let workerB = 0 + + await cache.getOrSet('["w",7,"list",[1]]', async () => { + workerA++ + return 'a1' + }) + await cache.getOrSet('["w",7,"overview",0]', async () => { + workerA++ + return 'a2' + }) + await cache.getOrSet('["w",9,"list",[1]]', async () => { + workerB++ + return 'b1' + }) + + const evicted = cache.evictByPrefix('["w",7,') + assert.equal(evicted, 2, '应逐出打手 7 的两条，别的打手不受影响') + + // 再取：打手 7 的缓存已失效会重算，打手 9 直接命中 + await cache.getOrSet('["w",7,"list",[1]]', async () => { + workerA++ + return 'a1' + }) + await cache.getOrSet('["w",9,"list",[1]]', async () => { + workerB++ + return 'b1' + }) + assert.equal(workerA, 3, '打手 7 被逐出后应重新计算') + assert.equal(workerB, 1, '打手 9 未被逐出，应命中缓存') +}) + +test('evictByPrefix 对不匹配的前缀返回 0', () => { + const cache = new WorkerListCache(60_000) + assert.equal(cache.evictByPrefix('["w",999,'), 0) +}) + +test('evictByPrefix 会连并发合并中的在途 Promise 一起逐出', async () => { + const cache = new WorkerListCache(60_000) + let computed = 0 + + const first = cache.getOrSet('["w",7,"list",[1]]', async () => { + computed++ + return 'value' + }) + // 在途未完成时就逐出：条目应被移除，失败重试语义不受影响 + cache.evictByPrefix('["w",7,') + assert.equal(await first, 'value') + + await cache.getOrSet('["w",7,"list",[1]]', async () => { + computed++ + return 'value' + }) + assert.equal(computed, 2, '在途时被逐出后，下一次读取应重新计算') +}) diff --git a/apps/backend/src/services/worker-platform/worker-my-order-cache.ts b/apps/backend/src/services/worker-platform/worker-my-order-cache.ts new file mode 100644 index 00000000..9da61f16 --- /dev/null +++ b/apps/backend/src/services/worker-platform/worker-my-order-cache.ts @@ -0,0 +1,50 @@ +import { WorkerListCache } from './worker-list-cache.js' + +/** + * 打手端「我的订单」列表/概览的缓存归属模块。 + * + * 为什么单独一个文件：缓存实例需要同时被两方引用—— + * 1. worker-my-order-service（读，getOrSet）； + * 2. realtime-event-service（写，事件驱动失效）。 + * 而前者已经 import 了后者，把实例放在任何一方都会形成循环依赖， + * 所以抽成只依赖 WorkerListCache 的中立模块。 + * + * TTL 策略：配合 publishWorkOrderSnapshots 里的精确失效，把 TTL 从 2s/5s 提到 30s。 + * 失效在工单流转（含指派/拼单/取消申请/反馈处理等 21 处写入路径）时触发， + * 所以 30s 只是兜底——正常情况下数据一变缓存立刻作废，长 TTL 不会造成可见滞后； + * 漏掉失效的最坏情况是打手最多 30s 看到旧状态（原 5s TTL 同样存在这个窗口）。 + */ +const MY_ORDER_LIST_CACHE_TTL_MS = 30_000 +const MY_ORDER_OVERVIEW_CACHE_TTL_MS = 30_000 + +const myOrderListCache = new WorkerListCache(MY_ORDER_LIST_CACHE_TTL_MS) +const myOrderOverviewCache = new WorkerListCache(MY_ORDER_OVERVIEW_CACHE_TTL_MS) + +/** 统一前缀，让事件失效可以按打手整体逐出该打手的全部「我的订单」缓存。 */ +function buildMyOrderCacheKey(workerId: number | string, scope: string, detail: unknown): string { + return JSON.stringify(['w', Number(workerId) || 0, scope, detail]) +} + +/** + * 逐出指定打手的「我的订单」列表与概览缓存。 + * 在工单发生影响该打手视图的变更时调用（见 realtime-event-service 的快照广播）。 + * 返回逐出的条目数，便于测试与观测。 + */ +export function invalidateWorkerMyOrderCaches( + workerIds: Array, +): number { + const normalized = new Set( + workerIds.map((item) => Number(item || 0)).filter((item) => Number.isFinite(item) && item > 0), + ) + if (normalized.size === 0) return 0 + + let evicted = 0 + for (const workerId of normalized) { + const prefix = `["w",${workerId},` + evicted += myOrderListCache.evictByPrefix(prefix) + evicted += myOrderOverviewCache.evictByPrefix(prefix) + } + return evicted +} + +export { myOrderListCache, myOrderOverviewCache, buildMyOrderCacheKey } diff --git a/apps/backend/src/services/worker-platform/worker-my-order-service.ts b/apps/backend/src/services/worker-platform/worker-my-order-service.ts index fe46427c..68e57b4a 100644 --- a/apps/backend/src/services/worker-platform/worker-my-order-service.ts +++ b/apps/backend/src/services/worker-platform/worker-my-order-service.ts @@ -44,23 +44,19 @@ import { type WorkerSession, } from './worker-session-context-service.js' import { settleOverdueWorkOrders } from './worker-order-timeout-service.js' -import { WorkerListCache } from './worker-list-cache.js' - -const MY_ORDER_LIST_CACHE_TTL_MS = 2_000 -const myOrderListCache = new WorkerListCache(MY_ORDER_LIST_CACHE_TTL_MS) +import { + buildMyOrderCacheKey, + myOrderListCache, + myOrderOverviewCache, +} from './worker-my-order-cache.js' /** - * 「我的订单」概览（各状态订单数 + 预计/已结算金额）的短缓存。 - * - * 这条查询 32 天被调用 130 万次（每秒约 0.5 次），是数据库累计耗时第一的热路径； - * 迁移 084 的覆盖索引把它从最坏 4.5~9 秒压到秒级以下，但每次仍要扫一遍索引， - * 所以再按打手缓存 5 秒：页面轮询/重复刷新时直接命中，把"计算次数"和"访问次数"解耦。 - * + * 「我的订单」列表/概览的缓存实例与 TTL 策略见 worker-my-order-cache.ts： + * TTL 30s + 工单流转时按打手精确失效（realtime-event-service 里触发）， + * 正常情况数据一变缓存立刻作废，30s 只是漏失效时的兜底。 * 只缓存读取本身：会话校验与 settleOverdueWorkOrders（有写副作用）每次照常执行， - * 否则逾期结算会被跳过。副作用是打手自己的订单数/金额最多滞后 5 秒。 + * 否则逾期结算会被跳过。 */ -const MY_ORDER_OVERVIEW_CACHE_TTL_MS = 5_000 -const myOrderOverviewCache = new WorkerListCache(MY_ORDER_OVERVIEW_CACHE_TTL_MS) export async function listWorkerMyOrders(query: JsonObject = {}, session: WorkerSession) { requireActiveWorkerSession(session) @@ -71,8 +67,7 @@ export async function listWorkerMyOrders(query: JsonObject = {}, session: Worker const permissions = resolveWorkerPermissions(worker) const statuses = normalizeStatuses(query.status) const keyword = String(query.keyword || '').trim() - const cacheKey = JSON.stringify([ - worker.id, + const cacheKey = buildMyOrderCacheKey(worker.id, 'list', [ page, pageSize, statuses, @@ -143,8 +138,9 @@ export async function getWorkerMyOrderOverview(session: WorkerSession) { requireActiveWorkerSession(session) const worker = await getRequiredWorker(session.workerId) await settleOverdueWorkOrders({ workerId: worker.id, limit: 50 }) - const rows = await myOrderOverviewCache.getOrSet(String(worker.id), () => - getWorkerOrderOverview(worker.id), + const rows = await myOrderOverviewCache.getOrSet( + buildMyOrderCacheKey(worker.id, 'overview', 0), + () => getWorkerOrderOverview(worker.id), ) const items = rows.map((row) => ({ status: row.status,