一次「补发把订单事件补丢了」的复盘
「回声」的端上曾经用一个简单的失败重发：请求失败就把事件塞回数组，下次启动时全部再发一遍。平时没问题，弱网大促时出了两类事故。
第一类是把队列撑爆。首页曝光事件量最大，断网十分钟就积压几千条，恢复后一次性并发上报，把端上内存和网络都打满，真正重要的支付事件排在最老的位置，反而被服务端限流拒掉。
第二类是重复。重发时没有保留原 eventId，每次重新生成，服务端无法判断「这是重试」还是「用户又买了一次」，订单事件被重复计数。
复盘结论很直接：队列不是数组加定时器，它要同时处理批量、重试、上限、丢弃、持久化和去重六件事。 这六件事互相牵制，单独优化任何一件都会把问题推到另一件上。
队列要回答的六个问题
| 问题 | 可选答案 | 默认取舍 |
|---|---|---|
| 一次发多少 | 单条、固定批量、按字节预算 | 固定批量，按时间窗触发 |
| 失败怎么重试 | 立即重试、固定间隔、指数退避 | 指数退避，设最大次数 |
| 队列多长 | 无上限、按条数、按字节 | 按条数，防止内存无界 |
| 满了丢谁 | 丢最旧、丢最低优先级、拒新 | 丢最低优先级，保护订单事件 |
| 进程退出怎么办 | 不持久化、持久化待发、持久化含在途 | 持久化待发与在途 |
| 重复怎么识别 | 服务端按内容去重、端上按 eventId | 端上先去重，服务端再兜底 |
这张表没有唯一正确答案，但每一项都必须显式写出来。最常见的故障不是选错策略，而是根本没意识到这里有一个策略。
优先级不是所有事件都一样
交易事件可以粗分三档：
critical：下单、支付、退款。丢了就无法对账，宁可挤掉别的。normal：加购、点击。影响分析，但不影响资金。low：曝光、页面浏览。量大，可采样、可丢弃。
队列满时按优先级淘汰，比「先进先出丢最旧」更符合交易场景。因为最旧的那条可能恰好是支付事件，而最新的一条只是首页曝光。反过来说，如果新来的事件优先级不高于将被淘汰的事件，就不应该为了它挤掉旧数据——这只会让队列在原地抖动。
一个内存加注入存储的队列
下面的模块属于 L2：队列状态在内存里，存储与时钟都从构造函数注入。测试用假时钟推进退避，用内存存储模拟崩溃恢复，不碰真实网络。
// file: src/analytics/report-queue.ts
// tier: L2
// verify: node --test test/analytics/report-queue.test.ts
export type QueuePriority = "low" | "normal" | "critical";
export type DropPolicy = "drop_lowest_priority" | "drop_oldest" | "reject_new";
export interface ReportEvent {
eventId: string;
name: string;
payload: string;
priority?: QueuePriority;
}
export interface QueuedEvent {
eventId: string;
name: string;
payload: string;
priority: QueuePriority;
enqueuedAt: number;
attempts: number;
nextAttemptAt: number;
inflightAt: number | null;
}
export interface PersistedQueue {
version: 1;
events: QueuedEvent[];
inflight: QueuedEvent[];
seen: string[];
deadLetters: QueuedEvent[];
}
export interface QueueStorage {
load(): PersistedQueue | null;
save(state: PersistedQueue): void;
}
export interface QueueOptions {
now: () => number;
storage: QueueStorage;
capacity?: number;
batchSize?: number;
maxAttempts?: number;
backoffMs?: number;
dropPolicy?: DropPolicy;
}
export interface EnqueueResult {
accepted: boolean;
reason: "ok" | "duplicate" | "queue_full";
droppedId: string | null;
}
export interface NackResult {
requeued: number;
dead: number;
}
const PRIORITY_RANK: Record<QueuePriority, number> = { low: 1, normal: 2, critical: 3 };
function copyEvent(event: QueuedEvent): QueuedEvent {
return { ...event };
}
function assertPositiveInt(value: number, name: string): void {
if (!Number.isInteger(value) || value < 1) throw new Error(`${name} 必须是正整数`);
}
export class ReportQueue {
pending: QueuedEvent[];
inflight: QueuedEvent[];
seen: Map<string, number>;
deadLetters: QueuedEvent[];
now: () => number;
storage: QueueStorage;
capacity: number;
batchSize: number;
maxAttempts: number;
backoffMs: number;
dropPolicy: DropPolicy;
dropped: number;
constructor(options: QueueOptions) {
this.pending = [];
this.inflight = [];
this.seen = new Map();
this.deadLetters = [];
this.now = options.now;
this.storage = options.storage;
this.capacity = options.capacity ?? 100;
this.batchSize = options.batchSize ?? 20;
this.maxAttempts = options.maxAttempts ?? 5;
this.backoffMs = options.backoffMs ?? 1_000;
this.dropPolicy = options.dropPolicy ?? "drop_lowest_priority";
this.dropped = 0;
assertPositiveInt(this.capacity, "capacity");
assertPositiveInt(this.batchSize, "batchSize");
assertPositiveInt(this.maxAttempts, "maxAttempts");
if (!Number.isInteger(this.backoffMs) || this.backoffMs < 0) {
throw new Error("backoffMs 必须是非负整数");
}
}
private persist(): void {
this.storage.save({
version: 1,
events: this.pending.map(copyEvent),
inflight: this.inflight.map(copyEvent),
seen: [...this.seen.keys()],
deadLetters: this.deadLetters.map(copyEvent),
});
}
private lowestPriorityIndex(): number {
let best = 0;
for (let i = 1; i < this.pending.length; i += 1) {
if (PRIORITY_RANK[this.pending[i].priority] < PRIORITY_RANK[this.pending[best].priority]) {
best = i;
}
}
return best;
}
enqueue(event: ReportEvent): EnqueueResult {
if (typeof event.eventId !== "string" || event.eventId.trim() === "") {
throw new Error("事件 id 不能为空");
}
if (typeof event.name !== "string" || event.name.trim() === "") {
throw new Error("事件名不能为空");
}
if (typeof event.payload !== "string" || event.payload === "") {
throw new Error("事件载荷不能为空");
}
const at = this.now();
if (this.seen.has(event.eventId)) {
return { accepted: false, reason: "duplicate", droppedId: null };
}
const queued: QueuedEvent = {
eventId: event.eventId,
name: event.name,
payload: event.payload,
priority: event.priority ?? "normal",
enqueuedAt: at,
attempts: 0,
nextAttemptAt: at,
inflightAt: null,
};
if (this.pending.length >= this.capacity) {
if (this.dropPolicy === "reject_new") {
return { accepted: false, reason: "queue_full", droppedId: null };
}
const victimIndex = this.dropPolicy === "drop_oldest" ? 0 : this.lowestPriorityIndex();
const victim = this.pending[victimIndex];
if (
this.dropPolicy === "drop_lowest_priority" &&
PRIORITY_RANK[queued.priority] <= PRIORITY_RANK[victim.priority]
) {
return { accepted: false, reason: "queue_full", droppedId: null };
}
this.pending.splice(victimIndex, 1);
this.dropped += 1;
this.pending.push(queued);
this.seen.set(queued.eventId, at);
this.persist();
return { accepted: true, reason: "ok", droppedId: victim.eventId };
}
this.pending.push(queued);
this.seen.set(queued.eventId, at);
this.persist();
return { accepted: true, reason: "ok", droppedId: null };
}
drainBatch(at: number = this.now()): QueuedEvent[] {
const ready: QueuedEvent[] = [];
const rest: QueuedEvent[] = [];
for (const event of this.pending) {
if (ready.length < this.batchSize && event.nextAttemptAt <= at) {
event.inflightAt = at;
ready.push(event);
} else {
rest.push(event);
}
}
if (ready.length === 0) return [];
this.pending = rest;
for (const event of ready) this.inflight.push(event);
this.persist();
return ready;
}
ack(ids: string[]): number {
const wanted = new Set(ids);
const before = this.inflight.length;
this.inflight = this.inflight.filter((event) => !wanted.has(event.eventId));
const removed = before - this.inflight.length;
if (removed > 0) this.persist();
return removed;
}
nack(ids: string[], retryable = true): NackResult {
const wanted = new Set(ids);
const at = this.now();
const moving = this.inflight.filter((event) => wanted.has(event.eventId));
this.inflight = this.inflight.filter((event) => !wanted.has(event.eventId));
let requeued = 0;
let dead = 0;
for (const event of moving) {
event.attempts += 1;
event.inflightAt = null;
if (!retryable || event.attempts >= this.maxAttempts) {
this.deadLetters.push(event);
dead += 1;
} else {
event.nextAttemptAt = at + this.backoffMs * 2 ** (event.attempts - 1);
this.pending.push(event);
requeued += 1;
}
}
if (moving.length > 0) this.persist();
return { requeued, dead };
}
// 进程崩溃时 in-flight 可能已发出但未确认，恢复时一律重排，由 eventId 在服务端幂等。
restore(): number {
const state = this.storage.load();
if (state === null) return 0;
if (state.version !== 1) throw new Error("队列存储版本不支持");
const at = this.now();
this.pending = state.events.map(copyEvent);
this.inflight = [];
this.deadLetters = state.deadLetters.map(copyEvent);
this.seen = new Map(state.seen.map((id) => [id, at]));
for (const event of state.inflight) {
const copy = copyEvent(event);
copy.inflightAt = null;
copy.nextAttemptAt = at;
this.pending.push(copy);
}
this.persist();
return this.pending.length;
}
stats(): { pending: number; inflight: number; seen: number; deadLetters: number; dropped: number } {
return {
pending: this.pending.length,
inflight: this.inflight.length,
seen: this.seen.size,
deadLetters: this.deadLetters.length,
dropped: this.dropped,
};
}
pendingIds(): string[] {
return this.pending.map((event) => event.eventId);
}
inflightIds(): string[] {
return this.inflight.map((event) => event.eventId);
}
deadIds(): string[] {
return this.deadLetters.map((event) => event.eventId);
}
nextReadyAt(): number | null {
if (this.pending.length === 0) return null;
let next = this.pending[0].nextAttemptAt;
for (const event of this.pending) {
if (event.nextAttemptAt < next) next = event.nextAttemptAt;
}
return next;
}
}
seen 与 pending 是两套状态。 seen 记录「这个 eventId 已经进过队列」，确认送达后也不删除；pending 只记录「还没发出去」。如果只靠 pending 去重，事件一旦发出，同一个 id 就能再次入队。
drainBatch 只移入 inflight，不删除。 只有 ack 才代表服务端接受。发送失败调用 nack，事件带着递增的 attempts 回到 pending，退避时间按 2 ** (attempts - 1) 增长。
崩溃恢复一律重排。 进程退出时 inflight 里的事件可能已经到达服务端，也可能没有。端上无法区分，所以恢复时全部放回 pending，依靠 eventId 在服务端做幂等。这是「至少一次」投递，不是「恰好一次」。
丢弃不写 seen。 被挤掉的事件没有送达，如果把它标记为已见，后续重试会被误判为重复。丢弃就是丢弃，让上游的采样或补偿逻辑去处理。
测试与真实输出
本机运行结果如下：
$ node --test --test-concurrency=1 test/analytics/report-queue.test.ts
✔ 入队后可按批量大小取出 (1.2535ms)
✔ 重复 eventId 被拒绝且不影响原事件 (0.102ms)
✔ 确认送达后重复 eventId 仍被拒绝 (0.119583ms)
✔ 队列满时优先丢弃低优先级事件 (0.132292ms)
✔ 新事件优先级不高于受害者时拒绝入队 (0.094833ms)
✔ drop_oldest 策略总是丢弃最旧事件 (0.073708ms)
✔ reject_new 策略在队列满时拒绝新事件 (0.076792ms)
✔ 未到重试时间的事件不会被取出 (0.175792ms)
✔ 重试退避按次数翻倍 (0.115291ms)
✔ 达到最大尝试次数后进入死信 (0.135ms)
✔ 不可重试错误直接进入死信 (0.08ms)
✔ 待发事件与去重集合可以持久化恢复 (0.125541ms)
✔ 恢复时未确认的 in-flight 会重新入队 (0.072833ms)
✔ 每次状态变更都会写入存储 (0.052416ms)
✔ 非法事件与非法容量被拒绝 (0.262833ms)
ℹ tests 15
ℹ suites 0
ℹ pass 15
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 152.886459
十五条用例覆盖：批量取件、重复拒绝、确认后仍去重、低优先级淘汰、同级拒绝、两种丢弃策略、退避等待、指数增长、最大次数、不可重试、持久化恢复、在途重排、写盘次数、非法输入。
一次「重复上报把 GMV 翻倍」的复盘
某交易客户端在支付回跳页上报 trade.order_pay。上报失败时，队列按指数退避重试三次。问题出在 eventId 的生成位置：它在每次发送前生成，而不是在事件产生时生成。
于是一笔支付在弱网下被上报了三次，服务端看到三个不同的 id，按三条支付记录入账。大盘 GMV 在一天内明显偏高，排查时却发现订单表没有对应增量。
修复只有一行：eventId 在事件创建时生成并随队列项一起持久化，重试复用同一个值。服务端同时补了一层按 eventId 的幂等表，即使端上逻辑回退，也不会重复入账。
离线缓存与队列的边界
一个可执行的划分是：
- 队列只保存待上报的事件，有明确的上限与淘汰策略，消费后即删除。
- 缓存保存的是可重放的业务数据，比如商品详情、配置、草稿，有 TTL 和版本号。
- 两者都不要无限增长。端上磁盘配额有限，超过阈值时应先降级低优先级事件，而不是让整个存储写失败。
小结
上报队列要同时回答六个问题：批量、重试、上限、丢弃、持久化、去重。交易场景的默认取舍是固定批量、指数退避、按条数限容、按优先级淘汰、持久化待发与在途、按 eventId 去重。
ReportQueue 把 seen 与 pending 分开，ack 之后才删除，崩溃恢复一律重排，把「至少一次」投递的责任交给服务端幂等兜底。
阅读导航：上一节：12.1 埋点模型与事件契约 · 下一节：12.3 漏斗与归因的端上边界 。
常见错误与边界
- 重试时重新生成
eventId，服务端把一次支付记成多笔。 - 队列无上限，弱网积压后一次性并发上报，把重要事件挤到限流队列末尾。
- 用「先进先出丢最旧」淘汰，最先丢掉的恰好是支付事件。
- 发送成功就删除事件，服务端实际未接受，数据静默丢失。
ack之前就写seen，发送失败后无法重试。- 进程退出不持久化在途事件，重启后这批数据永久消失。
- 把业务缓存与上报队列塞进同一个存储，TTL 与淘汰策略互相打架。
- 事件体原样落盘，手机号、地址等敏感字段进入本地明文存储。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。