轻应用海量埋点上报：批量聚合、背压控制与失败重传的高吞吐设计

class Tracker {
  constructor({ maxSize = 20, interval = 5000, sender }) {
    this.queue = [];
    this.maxSize = maxSize;
    this.sender = sender;
    this.timer = setInterval(() => this.flush(), interval);
    // 页面切后台/关闭前尽力发一次
    document.addEventListener('visibilitychange', () => {
      if (document.visibilityState === 'hidden') this.flush(true);
    });
  }

  track(event) {
    this.queue.push({
      ...event,
      ts: Date.now(),
      eventId: genUuid(),   // 每条事件唯一ID，服务端据此去重
    });
    if (this.queue.length >= this.maxSize) this.flush();
  }

  async flush(useBeacon = false) {
    if (this.queue.length === 0) return;
    const batch = this.queue.splice(0, this.queue.length);
    if (useBeacon && navigator.sendBeacon) {
      // 页面卸载场景用 sendBeacon，浏览器保证尽力发出
      navigator.sendBeacon('/collect', new Blob([JSON.stringify({ events: batch })]));
      return;
    }
    await this.sender.send(batch);
  }
}

enqueueWithBackpressure(event) {
  if (this.queue.length >= this.hardLimit) {
    // 背压：优先保留关键事件（下单/支付），丢弃高频低价值事件（曝光/滑动）
    if (event.priority === 'LOW') {
      this.droppedCount++;
      return;
    }
    this.queue.shift(); // 高优先级事件顶掉最老的一条
  }
  this.queue.push(event);
}

// 发送层：失败时指数退避，限制并发，避免弱网下请求越堆越多
async function sendWithBackoff(batch, retry = 0) {
  if (retry > 5) return persistToLocal(batch); // 重传失败转本地持久化
  try {
    await fetch('/collect', { method: 'POST', body: JSON.stringify({ events: batch }), keepalive: true });
  } catch (e) {
    await sleep(2 ** retry * 1000 + Math.random() * 500);
    return sendWithBackoff(batch, retry + 1);
  }
}

async function persistToLocal(batch) {
  await idbAdd('pending_events', batch); // 落库，页面重启也不丢
}

window.addEventListener('online', async () => {
  let pending;
  while ((pending = await idbTake('pending_events', 50)).length) {
    try {
      await postBatch(pending);
      await idbRemove('pending_events', pending.map(e => e.eventId));
    } catch (e) { break; } // 仍失败就停下，等下一次 online
  }
});

import json

def collect(request):
    payload = json.loads(request.body)
    events = payload.get('events', [])
    if len(events) > 100:                      # 单批上限，防异常大包
        return reject()
    valid = [e for e in events if valid_schema(e)]
    # 投入MQ削峰，不同事件按用户ID分区，保证单用户有序
    mq.send_batch('event_stream', valid, partition_key=lambda e: e['uid'])
    return {'code': 0}

def consume_and_dedup(batch):
    # 用事件唯一ID幂等去重，重传不重复计数
    fresh = redis.sadd_batch('seen_event', [e['eventId'] for e in batch])
    warehouse.batch_insert([e for e, ok in zip(batch, fresh) if ok])