perf(全链路): 连接池/事务边界/轮询与 IO 效率收口

Java:
- LLM 180s 读超时不再被全局 call-timeout 静默截断成 90s(长思考请求被掐断→重试→付费网关二次计费)
- 代理 HttpClient 缓存改有界 LRU(jikip 每次提取新 IP,无界缓存持续泄漏 selector 线程与连接池)
- 12 个 service 的 Redis 任务锁移出 @Transactional(自旋最坏 10s 白占 DB 连接,池仅 30),远端对象删/传改 afterCommit
- 结果文件 Job 闸门拒绝时不再回退内联执行(改重新入队,避免把背压转嫁给 MQ 消费线程)
- imagevideo 每秒扫描加列投影、过期清理加 LIMIT;权限页整表查询改列投影(不再拉回密码哈希)
- 哈希改 HexFormat;补 5 处"不能改"的技术依据注释(批量插入会丢回填主键、流式丢模板与图片等)

前端:4 个工具页轮询改轻量端点(带 fallback);PriceTrack 快照节流写盘;候选店铺表分页;页面隐藏时停表

客户端:HTTP 连接池按出口复用(Session 仍每请求新建,保持无跨请求状态);品牌检测 WIPO 逐请求握手;
代理配置按 mtime 缓存;串行任务改专属池;异常降级为标签页重连;紫鸟启动改端口轮询;模板编译缓存;
日志上报连接与落盘收口;Flask 版本 API 改按请求复用连接
This commit is contained in:
2026-09-15 23:01:59 +08:00
parent d6f8368493
commit 228d481211
44 changed files with 1550 additions and 423 deletions
@@ -194,23 +194,29 @@ async function pollOnce() {
for (const key of Object.keys(lineProgressMap.value)) {
if (!busy.some((b) => String(b.id) === key)) delete lineProgressMap.value[key]
}
for (const item of busy) {
const id = Number(item.id)
if (!id) continue
try {
const res = await getBrandTask(id)
const lp = res?.line_progress
if (lp?.has_progress && lp.info) {
lineProgressMap.value[String(id)] = {
current: Number(lp.info.current_line) || 0,
total: Number(lp.info.total_lines) || 0,
phase: lp.info.phase,
}
} else {
delete lineProgressMap.value[String(id)]
// 并发拉取:此前在 for 循环里逐个 await,任务多时一轮耗时随任务数线性叠加
//(brand 模块没有批量进度端点,见 shared/api/endpoints.ts,只能逐个查,故用并发而非 batch)
const busyIds = busy.map((item) => Number(item.id)).filter((id) => Number.isFinite(id) && id > 0)
const details = await Promise.all(
busyIds.map(async (id) => {
try {
return { id, res: await getBrandTask(id) }
} catch {
// 单条详情失败不中断整轮轮询
return { id, res: null }
}
} catch {
// 单条详情失败不中断整轮轮询
}),
)
for (const { id, res } of details) {
const lp = res?.line_progress
if (lp?.has_progress && lp.info) {
lineProgressMap.value[String(id)] = {
current: Number(lp.info.current_line) || 0,
total: Number(lp.info.total_lines) || 0,
phase: lp.info.phase,
}
} else {
delete lineProgressMap.value[String(id)]
}
}
scheduleNextPoll()
@@ -179,6 +179,7 @@ import {
type CollectDataTaskDetailVo,
type UploadFileVo,
} from '@/shared/api/java-modules'
import { getPollingProgressBatch } from '@/shared/api/task-progress-polling.ts'
import { formatDateTime } from '@/shared/utils/datetime'
import { uploadPathsToJava } from '@/shared/utils/upload-to-java'
import { runBatchDelete } from '@/shared/utils/batch-delete'
@@ -236,24 +237,70 @@ const taskSnapshots = ref<Record<number, CollectDataTaskDetailVo>>({})
const POLLING_STORAGE_KEY = 'brand:collect-data:polling-task-ids'
/** 轮询可比对的轻量字段(/tasks/progress/light 白名单里本模块快照有的字段),只有它们变化才重建快照对象 */
function taskProgressSignatureOf(detail: CollectDataTaskDetailVo) {
const task = detail.task
if (!task) return ''
return [task.status ?? '', task.updatedAt ?? ''].join('|')
}
/**
* currentItems 的 taskId → 下标索引。
* 轮询每轮都要按 taskId 定位行,此前每次都对整份列表 map 一遍;有索引后只改命中的那一行。
* 列表被整体替换(loadHistory)时必须重建;使用处还会做一次 taskId 校验兜底。
*/
let currentItemsIndex = new Map<number, number>()
function rebuildCurrentItemsIndex() {
const next = new Map<number, number>()
currentItems.value.forEach((item, index) => {
const taskId = Number(item.taskId)
if (Number.isFinite(taskId) && taskId > 0) next.set(taskId, index)
})
currentItemsIndex = next
}
/** 原地更新 currentItems 里指定任务的那一行(找不到就什么都不做) */
function patchCurrentItem(
taskId: number,
latestItem: CollectDataHistoryItem | undefined,
status: string | undefined,
) {
let index = currentItemsIndex.get(taskId)
if (index === undefined || Number(currentItems.value[index]?.taskId) !== taskId) {
rebuildCurrentItemsIndex()
index = currentItemsIndex.get(taskId)
}
if (index === undefined) return
const current = currentItems.value[index]
if (!current) return
currentItems.value[index] = {
...current,
...(latestItem || {}),
taskStatus: status ?? latestItem?.taskStatus ?? current.taskStatus,
}
}
const progressLoop = useTaskProgressLoop<CollectDataTaskDetailVo>({
scope: 'collect-data-tab',
storageKey: POLLING_STORAGE_KEY,
fetchProgress: (ids) => getCollectDataTaskProgressBatch(ids),
// F5:主轮询只读 status/fileReady 等轻量白名单字段,走 light 端点减小响应体;
// 轻量端点异常/空结果时适配器自动回退重型 batch(否则会漏刷新结果)
fetchProgress: (ids) => getPollingProgressBatch('collectData', ids, {
fallback: () => getCollectDataTaskProgressBatch(ids),
}),
extractTaskId: (detail) => detail.task?.id ?? null,
extractStatus: (detail) => detail.task?.status ?? '',
onUpdate: (taskId, detail) => {
const prev = taskSnapshots.value[taskId]
taskSnapshots.value = {
...taskSnapshots.value,
[taskId]: prev
? { ...prev, task: { ...(prev.task || {}), ...(detail.task || {}) }, items: detail.items ?? prev.items }
: detail,
const merged: CollectDataTaskDetailVo = prev
? { ...prev, task: { ...(prev.task || {}), ...(detail.task || {}) }, items: detail.items ?? prev.items }
: detail
// 只有轮询字段(或行明细)真的变了才展开快照对象:整份快照含行明细,无变化时重建纯属浪费
if (!prev || merged.items !== prev.items || taskProgressSignatureOf(prev) !== taskProgressSignatureOf(merged)) {
taskSnapshots.value = { ...taskSnapshots.value, [taskId]: merged }
}
const latestItem = detail.items?.[0]
currentItems.value = currentItems.value.map((item) => item.taskId === taskId
? { ...item, ...(latestItem || {}), taskStatus: detail.task?.status ?? latestItem?.taskStatus ?? item.taskStatus }
: item)
patchCurrentItem(taskId, detail.items?.[0], detail.task?.status)
},
onTerminal: async (taskId) => {
if (taskSnapshots.value[taskId]) {
@@ -624,6 +671,8 @@ async function loadHistory() {
}
}
currentItems.value = merged
// 列表整体替换后重建 taskId 索引,轮询就能原地更新命中的那一行
rebuildCurrentItemsIndex()
} catch {
/* 忽略:历史接口失败时保持空列表 */
}
@@ -33,7 +33,7 @@
</div>
<div v-else class="candidate-table-scroll">
<el-table
:data="candidates"
:data="pagedCandidates"
row-key="id"
height="260"
class="candidate-table"
@@ -59,6 +59,14 @@
</el-table-column>
</el-table>
</div>
<el-pagination
v-if="candidateTotal > candidatePageSize"
class="candidate-pagination"
layout="total, prev, pager, next"
:total="candidateTotal"
:page-size="candidatePageSize"
v-model:current-page="candidatePage"
/>
<div class="section-title condition-title">
<span>删除条件</span>
@@ -310,6 +318,13 @@ const {
total: matchedTotal,
paged: pagedMatchedItems,
} = useTablePaging(matchedItems)
// F9:备选店铺表同样只切渲染窗口(分页不改变数据源,勾选/删除语义不变)
const {
page: candidatePage,
pageSize: candidatePageSize,
total: candidateTotal,
paged: pagedCandidates,
} = useTablePaging(candidates)
const historyItems = ref<PatrolDeleteHistoryItem[]>([]);
const dashboard = ref<PatrolDeleteDashboardVo>({
candidateCount: 0,
@@ -467,10 +467,37 @@ function loadTaskSnapshotsFromStorage() {
}
}
function saveTaskSnapshotsToStorage() {
/** 快照落盘最小间隔:快照可达数百 KB,全量 JSON.stringify + setItem 会阻塞主线程 */
const SNAPSHOT_SAVE_MIN_INTERVAL_MS = 30000
let lastSnapshotSaveAt = 0
let snapshotSavePending = false
/**
* 快照写盘(节流)。
*
* 每轮轮询都序列化整份快照会周期性卡顿主线程,因此:
* - 内容未变时 setStorageJson 内部已会跳过写入;
* - 这里再按 SNAPSHOT_SAVE_MIN_INTERVAL_MS 节流,被节流掉的写入挂起,
* 下一次(或 force / 组件卸载前的 flush)落盘时一并写入最新内容。
*/
function saveTaskSnapshotsToStorage(force = false) {
if (typeof window === 'undefined') return
const now = Date.now()
if (!force && lastSnapshotSaveAt > 0 && now - lastSnapshotSaveAt < SNAPSHOT_SAVE_MIN_INTERVAL_MS) {
snapshotSavePending = true
return
}
snapshotSavePending = false
lastSnapshotSaveAt = now
setStorageJson(taskSnapshotsStorageKey(), taskSnapshots.value, Object.keys(taskSnapshots.value).length === 0)
}
/** 卸载前把节流挂起的快照补写一次,避免最后一次变更丢失 */
function flushTaskSnapshotsToStorage() {
if (!snapshotSavePending) return
saveTaskSnapshotsToStorage(true)
}
function loadMatchedItemsFromStorage() {
try {
const raw = typeof window !== 'undefined' ? window.localStorage.getItem(matchedItemsStorageKey()) : null
@@ -1147,7 +1174,7 @@ async function processMatchedQueue() {
...taskDetails.value,
[taskVo.taskId]: 'RUNNING',
}
saveTaskSnapshotsToStorage()
saveTaskSnapshotsToStorage(true)
saveTaskDetailsToStorage()
const queuePayload = await buildQueuePayload(taskVo, row)
queuePayloadText.value = JSON.stringify(queuePayload, null, 2)
@@ -1233,7 +1260,7 @@ function recordCreatedTask(taskVo: PriceTrackCreateTaskVo) {
...taskDetails.value,
[taskVo.taskId]: 'RUNNING',
}
saveTaskSnapshotsToStorage()
saveTaskSnapshotsToStorage(true)
saveTaskDetailsToStorage()
}
@@ -1418,37 +1445,68 @@ function getPollIntervalMs() {
return getTaskPollIntervalMs()
}
/** 轮询可比对的轻量字段(与 /tasks/progress/light 白名单一致,外加只有重型端点才带的起止时间) */
function taskProgressSignature(task?: PriceTrackTaskDetailVo['task'] | null) {
if (!task) return ''
return [
task.status ?? '',
task.createdAt ?? '',
task.finishedAt ?? '',
task.updatedAt ?? '',
].join('|')
}
async function refreshTaskBatch() {
const ids = pollingTaskIds.value.filter((id) => id > 0)
if (!ids.length) return
try {
const batch = await getPriceTrackTaskProgressBatch(ids)
let changed = false
// F5:主轮询只读 status/fileReady 等轻量白名单字段,走 light 端点减小轮询响应体;
// 轻量端点异常或返回空结果时适配器自动回退重型 batch(否则会漏刷新结果)
const batch = await getPollingProgressBatch('priceTrack', ids, {
fallback: () => getPriceTrackTaskProgressBatch(ids),
})
let settled = false
let snapshotsChanged = false
const nextSnapshots = { ...taskSnapshots.value }
for (const missingId of batch.missingTaskIds || []) {
console.log(`[price-track] 任务 ${missingId} 在服务端已不存在,停止轮询`)
removePollingTask(missingId)
delete nextSnapshots[missingId]
snapshotsChanged = true
}
for (const detail of batch.items || []) {
const taskId = detail.task?.id
const status = detail.task?.status
if (typeof taskId !== 'number' || taskId <= 0) continue
const prev = nextSnapshots[taskId]
nextSnapshots[taskId] = prev ? { ...prev, task: { ...(prev.task || {}), ...(detail.task || {}) } } : detail
const merged: PriceTrackTaskDetailVo = prev
? { ...prev, task: { ...(prev.task || {}), ...(detail.task || {}) } }
: detail
// 只有轮询字段真的变了才重写快照:整份快照数百 KB,无变化时重建对象 + 落盘纯属浪费
if (!prev || taskProgressSignature(prev.task) !== taskProgressSignature(merged.task)) {
nextSnapshots[taskId] = merged
snapshotsChanged = true
}
if (status) {
taskDetails.value[taskId] = status
if (status === 'SUCCESS' || status === 'FAILED') {
console.log(`[price-track] 任务 ${taskId} 已到终态 ${status},移出轮询集合并刷新列表`)
removePollingTask(taskId)
changed = true
settled = true
}
}
}
taskSnapshots.value = nextSnapshots
saveTaskSnapshotsToStorage()
if (snapshotsChanged) {
taskSnapshots.value = nextSnapshots
saveTaskSnapshotsToStorage()
}
saveTaskDetailsToStorage()
await loadHistory()
syncPollingIdsWithHistory()
if (changed) {
// 历史列表只在任务落定(或快照真变化)时拉取:此前每 10s 无条件重查一次,纯属浪费
if (settled || snapshotsChanged) {
await loadHistory()
syncPollingIdsWithHistory()
}
if (settled) {
await loadDashboard()
void resumeLoopExecutionIfNeeded()
}
@@ -1508,7 +1566,8 @@ function removePollingTask(taskId: number) {
const next = { ...taskSnapshots.value }
delete next[taskId]
taskSnapshots.value = next
saveTaskSnapshotsToStorage()
// 终态清理属于关键变更,跳过节流立即落盘
saveTaskSnapshotsToStorage(true)
}
}
@@ -1679,6 +1738,8 @@ onUnmounted(() => {
clearSleepTimers()
timers.clearScope()
if (countryPrefSaveTimer) timers.clearTimer('preference-save', countryPrefSaveTimer)
// 补写被节流挂起的快照,避免最后一次变更丢失
flushTaskSnapshotsToStorage()
})
/**
@@ -33,7 +33,7 @@
</div>
<div v-else class="candidate-table-scroll">
<el-table
:data="candidates"
:data="pagedCandidates"
row-key="id"
height="260"
class="candidate-table"
@@ -59,6 +59,14 @@
</el-table-column>
</el-table>
</div>
<el-pagination
v-if="candidateTotal > candidatePageSize"
class="candidate-pagination"
layout="total, prev, pager, next"
:total="candidateTotal"
:page-size="candidatePageSize"
v-model:current-page="candidatePage"
/>
<ZiniaoVersionSetting v-model="ziniaoVersion" />
@@ -245,6 +253,13 @@ const {
total: matchedTotal,
paged: pagedMatchedItems,
} = useTablePaging(matchedItems)
// F9:备选店铺表同样只切渲染窗口(分页不改变数据源,勾选/删除语义不变)
const {
page: candidatePage,
pageSize: candidatePageSize,
total: candidateTotal,
paged: pagedCandidates,
} = useTablePaging(candidates)
const historyItems = ref<QueryAsinHistoryItem[]>([]);
const dashboard = ref<QueryAsinDashboardVo>({
candidateCount: 0,
@@ -17,7 +17,7 @@
<div class="section-title">备选区</div>
<div v-if="!candidates.length" class="empty-candidates">暂无备选店铺</div>
<div v-else class="candidate-table-scroll">
<el-table :data="candidates" row-key="id" height="250" class="candidate-table"
<el-table :data="pagedCandidates" row-key="id" height="250" class="candidate-table"
@selection-change="onSelectionChange">
<el-table-column type="selection" width="42" />
<el-table-column prop="shop_name" label="店铺名" min-width="140" show-overflow-tooltip />
@@ -28,6 +28,9 @@
</el-table-column>
</el-table>
</div>
<el-pagination v-if="candidateTotal > candidatePageSize" class="candidate-pagination"
layout="total, prev, pager, next" :total="candidateTotal" :page-size="candidatePageSize"
v-model:current-page="candidatePage" />
<div class="section-title">抓取国家与顺序</div>
<div class="country-pref-checks">
@@ -187,6 +190,13 @@ const {
total: matchedTotal,
paged: pagedMatchedItems,
} = useTablePaging(matchedItems)
// F9:备选店铺表同样只切渲染窗口(分页不改变数据源,勾选/删除语义不变)
const {
page: candidatePage,
pageSize: candidatePageSize,
total: candidateTotal,
paged: pagedCandidates,
} = useTablePaging(candidates)
const historyItems = ref<ShopDataCrawlHistoryItem[]>([])
const dashboard = ref<ShopDataCrawlDashboardVo>({ candidateCount: 0, processedTaskCount: 0, successTaskCount: 0, failedTaskCount: 0 })
const orderedCountryCodes = ref<string[]>(COUNTRY_OPTIONS.map((row) => row.code))
@@ -452,7 +452,8 @@ function sleep(ms: number) { if (disposed) return Promise.resolve(); return time
function clearSleepTimers() { timers.clearCategory('queue-wait') }
function isTransientBackendError(error: unknown) { const message = (error instanceof Error ? error.message : String(error || '')).toLowerCase(); return TRANSIENT_BACKEND_ERROR_PATTERNS.some((pattern) => message.includes(pattern.toLowerCase())) }
async function withTransientRetry<T>(action: () => Promise<T>, onRetry: (attempt: number, maxAttempts: number) => void, maxAttempts = 10) { let attempt = 0; while (true) { if (disposed) throw new Error('component disposed'); try { return await action() } catch (error) { attempt += 1; if (!isTransientBackendError(error) || attempt >= maxAttempts) throw error; onRetry(attempt, maxAttempts); await sleep(getPollIntervalMs()) } } }
async function refreshTaskBatch() { if (!pollingTaskIds.value.length) return { settledTaskIds: [] as number[] }; try { const batch = await getShopMatchTaskProgressBatch(pollingTaskIds.value); const nextSnapshots = { ...taskSnapshots.value }; const settledTaskIds = new Set<number>(); for (const missingId of batch.missingTaskIds || []) { removeTaskLocally(missingId); delete nextSnapshots[missingId]; settledTaskIds.add(missingId) } for (const detail of batch.items || []) { const taskId = detail.task?.id; if (!taskId) continue; const prev = nextSnapshots[taskId]; nextSnapshots[taskId] = prev ? { ...prev, task: { ...(prev.task || {}), ...(detail.task || {}) } } : detail; const status = detail.task?.status || ''; if (!status) continue; taskDetails.value[taskId] = status; if (isTaskTerminalStatus(status)) { removePollingTask(taskId); settledTaskIds.add(taskId); continue } if (status === 'SCHEDULED') { stopPollingTask(taskId); settledTaskIds.add(taskId) } } taskSnapshots.value = nextSnapshots; saveTaskSnapshotsToStorage(); saveTaskDetailsToStorage(); restoreScheduledDispatches(); return { settledTaskIds: Array.from(settledTaskIds) } } catch { return { settledTaskIds: [] as number[] } } }
async function refreshTaskBatch() { if (!pollingTaskIds.value.length) return { settledTaskIds: [] as number[] }; try { // F5:主轮询只读 status 等轻量白名单字段,走 light 端点减小响应体;轻量异常/空结果时回退重型 batch(否则会漏刷新结果)
const batch = await getPollingProgressBatch('shopMatch', pollingTaskIds.value, { fallback: () => getShopMatchTaskProgressBatch(pollingTaskIds.value) }); const nextSnapshots = { ...taskSnapshots.value }; const settledTaskIds = new Set<number>(); for (const missingId of batch.missingTaskIds || []) { removeTaskLocally(missingId); delete nextSnapshots[missingId]; settledTaskIds.add(missingId) } for (const detail of batch.items || []) { const taskId = detail.task?.id; if (!taskId) continue; const prev = nextSnapshots[taskId]; nextSnapshots[taskId] = prev ? { ...prev, task: { ...(prev.task || {}), ...(detail.task || {}) } } : detail; const status = detail.task?.status || ''; if (!status) continue; taskDetails.value[taskId] = status; if (isTaskTerminalStatus(status)) { removePollingTask(taskId); settledTaskIds.add(taskId); continue } if (status === 'SCHEDULED') { stopPollingTask(taskId); settledTaskIds.add(taskId) } } taskSnapshots.value = nextSnapshots; saveTaskSnapshotsToStorage(); saveTaskDetailsToStorage(); restoreScheduledDispatches(); return { settledTaskIds: Array.from(settledTaskIds) } } catch { return { settledTaskIds: [] as number[] } } }
function getPollIntervalMs() { return getTaskPollIntervalMs() }
function scheduleNextPoll(immediate = false) { if (disposed) return; if (pollTimer.value) { if (!immediate) return; timers.clearTimer('task-poll', pollTimer.value); pollTimer.value = null } const run = async () => { pollTimer.value = null; if (disposed) return; if (pollingInFlight.value || !pollingTaskIds.value.length) { if (pollingTaskIds.value.length) scheduleNextPoll(); return } pollingInFlight.value = true; try { const { settledTaskIds } = await refreshTaskBatch(); if (settledTaskIds.length) await Promise.allSettled([loadHistory(), loadDashboard()]) } finally { pollingInFlight.value = false } if (!disposed && pollingTaskIds.value.length) pollTimer.value = timers.setTimeout('task-poll', run, getPollIntervalMs()) }; if (immediate) void run(); else pollTimer.value = timers.setTimeout('task-poll', run, getPollIntervalMs()) }
function ensurePolling(immediate = false) { scheduleNextPoll(immediate) }
@@ -44,7 +44,7 @@
</div>
<div v-else class="candidate-table-scroll">
<el-table
:data="candidates"
:data="pagedCandidates"
row-key="id"
height="260"
class="candidate-table"
@@ -70,6 +70,14 @@
</el-table-column>
</el-table>
</div>
<el-pagination
v-if="candidateTotal > candidatePageSize"
class="candidate-pagination"
layout="total, prev, pager, next"
:total="candidateTotal"
:page-size="candidatePageSize"
v-model:current-page="candidatePage"
/>
<ZiniaoVersionSetting v-model="ziniaoVersion" />
@@ -296,6 +304,13 @@ const {
total: matchedTotal,
paged: pagedMatchedItems,
} = useTablePaging(matchedItems)
// F9:备选店铺表同样只切渲染窗口(分页不改变数据源,勾选/删除语义不变)
const {
page: candidatePage,
pageSize: candidatePageSize,
total: candidateTotal,
paged: pagedCandidates,
} = useTablePaging(candidates)
const historyItems = ref<WithdrawHistoryItem[]>([]);
const dashboard = ref<WithdrawDashboardVo>({
candidateCount: 0,