构建 Server-Sent Events 广播服务

项目概述

本项目实现 Server-Sent Events 广播服务：向浏览器流式推送事件，断线后按最后事件 id 续传，并在不拖住广播者的前提下丢弃慢客户端。SSE 是最简单的推送方式，真正会坏的是重连与慢客户端。

架构与关键决策

客户端按频道订阅；广播者为每频道维护单调 id，使每个事件都有可回传的 id，并保留小环形缓冲用于重放。响应关闭缓冲并周期发送心跳注释，避免中间层关闭看似空闲的流。每个订阅者队列有界，填满即关闭连接，使一个慢客户端既不能阻塞广播者也不能无界增长内存。id 由客户端回传时视为不可信值并夹取到缓冲范围，防止读取无关历史。

实施与测试

连接客户端、广播事件断言收到；断开并以最后事件 id 重连断言补发遗漏事件；填满订阅者队列断言连接被关闭而非广播者阻塞；断言 id 超出缓冲时提示重新同步而非部分重放。

限制与后续

• 重放受内存环形缓冲限制，长时间断线需全量重同步。
• 投递不持久，断线期间落到缓冲外的事件会丢失。
• 跨实例无逐主题顺序保证，仅共享通道的投递顺序。

代码示例

pnpm init
pnpm add express
pnpm add -D typescript tsx vitest

sse-service/
├── src/stream.ts
├── src/broadcast.ts
├── src/replay.ts
├── src/heartbeat.ts
├── tests/sse.test.ts
└── package.json

res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive' });

const id = ++channel.sequence;

res.write('id: ' + id + '\nevent: ' + name + '\ndata: ' + JSON.stringify(payload) + '\n\n');

const since = Number(req.headers['last-event-id'] ?? 0);
for (const e of channel.buffer.filter((e) => e.id > since)) res.write(format(e));

if (res.writableLength > MAX) return res.end();

pnpm vitest run