perf(realtime): 一次广播只序列化一次,不再按订阅者重复 stringify

抓包发现单个后台客户端 5 秒收到 981,776 字节 / 336 条事件(约 196KB/s、67 条/秒、
每条平均 2680 字节,最大 6418 字节)。事件构成:work_order.changed 34 条/秒、
order.changed 23 条/秒、admin_notification.changed 7.6 条/秒。

原广播实现在订阅者循环里逐个 `JSON.stringify(payload)` 并重建 SSE 模板串:
客户端越多,同一份载荷就被序列化越多次(后台/打手端都在线时更明显)。
改成先 `buildEventFrame()` 构造一次,循环里只写同一帧:

- broadcast() 只调用一次 buildEventFrame
- writeEvent()(ready 事件与断线补发)复用同一构造函数
- 行为不变,仅去掉重复序列化

受众过滤经核对本来就是正确的(admin 事件只发后台、worker 事件按 workerIds 过滤),
realtime_events 也有自我清理(每 100 条保留最近 500 条,实测 555 行/2MB),
故本次只动序列化这一处。

验证:后端 548 项单测全绿(新增 1 项守卫,禁止 broadcast 内再出现 JSON.stringify)。
This commit is contained in:
yml2213
2026-10-01 15:46:43 +08:00
parent 1f8252a6bc
commit 142c4ee141
2 changed files with 43 additions and 7 deletions
@@ -51,3 +51,26 @@ test('dev-mock/status 不能加 HTTP 缓存(它是可切换状态,缓存会
'dev-mock/status 缓存后,POST 切换开关再读会拿到旧值,界面看起来没生效',
)
})
test('实时广播必须一次序列化、多路复用(不能在每个订阅者循环里各 stringify 一次)', async () => {
const src = await readFile(
new URL('../services/realtime/realtime-event-service.ts', import.meta.url),
'utf8',
)
assert.ok(src.includes('function buildEventFrame('), '缺少预构造 SSE 帧的函数')
const broadcastBody = src.slice(
src.indexOf('function broadcast('),
src.indexOf('/** 预先构造一整帧 SSE 报文'),
)
assert.ok(broadcastBody.includes('const frame = buildEventFrame('), '广播必须先构造一次帧')
const broadcastCode = broadcastBody
.split('\n')
.filter((line) => !line.trim().startsWith('//'))
.join('\n')
assert.equal(
broadcastCode.includes('JSON.stringify'),
false,
'广播代码里不能出现 JSON.stringify:那意味着每个订阅者各序列化一次(生产实测 67 条/秒 × 客户端数)',
)
assert.ok(broadcastBody.includes('writeEventFrame(subscriber, frame)'), '循环里只应写入同一帧')
})
@@ -471,24 +471,37 @@ function isEventVisibleToSubscriber(event: StoredRealtimeEvent, subscriber: Real
}
function broadcast(subscribers: Iterable<RealtimeSubscriber>, event: RealtimeEvent) {
// 同一份载荷要发给 N 个订阅者,序列化必须只做一次。
// 生产实测单个后台客户端就能收到约 67 条/秒、每条约 2.7KB 的事件,
// 原实现在订阅者循环里各写一次 JSON.stringify + 模板串,客户端越多越吃亏。
const frame = buildEventFrame('realtime', event)
for (const subscriber of subscribers) {
writeEvent(subscriber, 'realtime', event)
writeEventFrame(subscriber, frame)
}
}
/** 预先构造一整帧 SSE 报文(含 `id:` 行),供广播复用。 */
function buildEventFrame(name: string, payload: unknown): string {
const eventId =
name === 'realtime' && payload && typeof payload === 'object' && 'eventId' in payload
? Number((payload as { eventId?: unknown }).eventId || 0)
: 0
const idLine = eventId > 0 ? `id: ${eventId}\n` : ''
return `${idLine}event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`
}
function writeEvent(subscriber: RealtimeSubscriber, name: string, payload: unknown) {
writeEventFrame(subscriber, buildEventFrame(name, payload))
}
function writeEventFrame(subscriber: RealtimeSubscriber, frame: string) {
if (subscriber.response.writableEnded || subscriber.response.destroyed) {
removeSubscriber(subscriber)
return
}
try {
const eventId =
name === 'realtime' && payload && typeof payload === 'object' && 'eventId' in payload
? Number((payload as { eventId?: unknown }).eventId || 0)
: 0
const idLine = eventId > 0 ? `id: ${eventId}\n` : ''
subscriber.response.write(`${idLine}event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`)
subscriber.response.write(frame)
} catch {
removeSubscriber(subscriber)
}