流的可靠性：速率限制（rate‑limits）、反压（backpressure）与事件监控

1. 为什么流对负载格外敏感

在前面的部分里，我们已经拆解了 MCP 事件、job.progress/job.completed 状态、异步 job 以及 GiftGenius 的流式通道（SSE/HTTP-stream）。现在要看的是：当它们遇到真实负载时会发生什么。

当你只有一个偶尔运行选礼流程的用户时，一切都很美好。但一旦 GiftGenius 上线生产，同时涌入上百个“给公司所有员工挑礼物”的请求，你会突然发现：

• 服务器上有数百条长寿命的 SSE 连接；
• worker 兴致勃勃地在每个细枝末节上发送 job.progress；
• 日志每天以 GB 规模增长；
• 用户端 UI 开始卡顿，尽管“服务器似乎并未宕机”。

经典 HTTP 请求只活几毫秒或几秒。SSE 或 HTTP‑stream 则可能活几分钟甚至数小时。它会占住连接、内存、文件描述符。每个发出的 event 都意味着 JSON 序列化、网络拷贝、GC 参与。如果把这当作“后端再多打一条 console.log”，系统很快就会变成“取暖器”。

MCP 事件还有一个特点：它们常常对同一任务被多次生成。一个每 0.1% 更新一次进度的 worker，会让每个 job 的事件数目惊人。结果就是“噪音”：海量细碎消息，它们：

• 拖累网络与 CPU；
• 塞满队列与缓冲区；
• 让调试与日志分析变得痛苦。

因此，对流与 MCP 事件的态度应与对数据库请求或模型调用一样严肃：它们是昂贵资源，需要配额、控制与监控。

为此，牢记三大主题：

1. Rate‑limits —— 限制我们能生成和发送事件/流的频次与数量。
2. Backpressure —— 当消费者跟不上生产者时的应对策略。
3. 监控与指标 —— 度量系统状态，及时发现“沸腾”的征兆。

2. 流与事件的速率限制（Rate‑limiting）

先从最直观的“限流”说起。

要明白，在流式场景里，“危险的违规者”往往不是客户端，而是服务器。在普通 REST‑API 中，你限制的是客户端对服务器的请求次数，避免用户把你 DDoS 掉。在 MCP 与流的世界里，则很容易发生反向 DDoS：worker 或 MCP 服务器每秒向客户端猛轰数千个事件。

需要哪些限制

通常从三个维度考虑。

其一，对用户或会话的限制。不能允许单个用户打开二十个 GiftGenius 主向导，每个都绑定一个 SSE 流。合理的限制是：每个会话仅几个活动流，并限制单个用户或租户处于 running 状态的 job 数量。

其二，对单个 job 的限制。这里关注事件频率。job.progress 完全可以设为每 N 毫秒最多发送一次，或者仅在有显著变化时发送，比如每 5% 的进度才发一条。没必要为目录里处理的每一件商品都发消息。同时要限制 payload 大小：进度事件不应携带数 MB 的文本。

其三，对 IP 或组织的限制。这是防滥用：有人跑脚本疯狂提交任务，或你的 App 突然爆火。此时可以让 API‑gateway 与代理的常见机制上场。

事件频率限制的简单实现

考虑 GiftGenius 的 worker：它在后台按照长名单为收礼人选礼，并周期性地通过 MCP‑notification event/progress 发送进度。我们希望事件发送不超过每 500 毫秒一次，且仅当百分比至少变化 5 个点时发送。

用于说明的 TS 伪代码：

// 假设我们有一个 mcpClient.sendNotification(...)
let lastSentPercent = 0;
let lastSentAt = 0;

function reportProgress(jobId: string, percent: number, message: string) {
  const now = Date.now();
  const percentDelta = percent - lastSentPercent;
  const timeDelta = now - lastSentAt;

  // 仅当已过去 >= 500 ms 或者百分比增长 >= 5% 时才发送
  if (percentDelta >= 5 || timeDelta >= 500) {
    mcpClient.sendNotification("event/progress", {
      jobId,
      percent,
      message,
    });
    lastSentPercent = percent;
    lastSentAt = now;
  }
}

这种方式叫“节流”（throttling）：我们按时间和数值变化对事件流进行“抽稀”。

如果你把流程拆成阶段（“第 1/3 阶段”“第 2/3 阶段”），逻辑更简单：只在阶段切换时发送事件。

限制同时打开的流数量

在 MCP 服务器一侧，你大概率有一个 SSE 的 HTTP 处理器：

// app/api/events/[userId]/route.ts (Next.js 16 App Router)
export async function GET(
  req: Request,
  { params }: { params: { userId: string } },
) {
  const userId = params.userId;

  if (!canOpenMoreStreams(userId)) {
    return new Response("Too many streams", { status: 429 });
  }

  const stream = new ReadableStream({
    start(controller) {
      registerSseClient(userId, controller);
    },
    cancel() {
      unregisterSseClient(userId);
    },
  });

  return new Response(stream, {
    headers: { "Content-Type": "text/event-stream" },
  });
}

函数 canOpenMoreStreams 可以检查某用户当前已打开的连接数并与阈值比较（例如最多 3 条并行流）。若超过限制，就返回 429，并在 GPT 指令里告知模型：此时不要再启动一个新的长向导，而是提醒用户“已有一个进行中的选礼流程，请先等待它完成”。

在小系统中，这类检查可以在进程内存中实现。在更严肃的基础设施里，它通常下沉到 MCP‑gateway 或独立的 rate‑limit 服务。

3. Backpressure：当消费者跟不上时该做什么

Rate‑limits 决定我们期望生产多少事件。即便限得很克制，也仍可能出现“消费者跟不上”的情况：用户的移动网络很差、浏览器标签页卡住了、ChatGPT 此刻极度繁忙。

Backpressure 是系统对“消费者处理不过来”的响应。与其无限堆积数据、最终因 OOM 崩溃，不如有意识地：

• 放慢速度；
• 聚合事件；
• 丢弃次要事件。

压力从哪里产生

GiftGenius 的典型场景是：worker 把事件写入队列（如 Redis Streams 或数据库表），MCP 服务器读取并推送到 SSE 通道。如果客户端很慢（3G、老旧笔记本、多标签页占用），TCP 缓冲开始积压，Node 进程无法及时清空队列，最终在内存里囤积事件。然后你会看到熟悉的一幕：

FATAL ERROR: Ineffective mark-compacts near heap limit

网络层（TCP）的 backpressure 已经存在，但它不了解你的领域对象。它只会说：“嘿，慢点，缓冲区满了。”我们的任务是把这种信号映射到 MCP 事件层面。

带上限的缓冲与事件丢弃

针对进度和状态，我们有个好处：并非所有事件价值相同。用户更在意“最新的百分比”，而不是完整的中间历史“51%、52%、53%、54%”。这意味着我们可以放心地丢弃部分事件，仅发送最新的一条。

假设有一层从 worker 接收进度事件，并按 jobId 把它们放入缓冲：

type ProgressEvent = { jobId: string; percent: number; message: string };

const progressBuffers = new Map<string, ProgressEvent[]>();
const MAX_BUFFER = 10;

function bufferProgress(event: ProgressEvent) {
  const buffer = progressBuffers.get(event.jobId) ?? [];
  buffer.push(event);

  // 限制缓冲区大小
  if (buffer.length > MAX_BUFFER) {
    // 只保留最后的若干条事件
    progressBuffers.set(event.jobId, buffer.slice(-MAX_BUFFER));
  } else {
    progressBuffers.set(event.jobId, buffer);
  }
}

单独的定时器（例如每 500 ms）查看缓冲，只发送最后一条，忽略其它：

setInterval(() => {
  for (const [jobId, buffer] of progressBuffers.entries()) {
    if (!buffer.length) continue;

    const last = buffer[buffer.length - 1];
    sendProgressToClient(last); // SSE/MCP notification

    progressBuffers.set(jobId, []); // 清空
  }
}, 500);

这是 conflation（合并）策略：把多个更新合成一条“当前真相”。对进度类事件来说是黄金模式。

对于“log”或 partial_result 之类的事件，策略可能不同。那里通常不允许丢失：日志文本很重要，丢一个 JSON 分片可能破坏数据结构。在这些情况下可以：

• 聚合消息（把多行日志粘成一包）；
• 或者向 worker 发送控制信号，让它“放慢日志生成”。

在异步系统中第二种更困难，但值得纳入考量。

限制队列深度

Backpressure 不只发生在发送前的事件缓冲，要审视系统中的所有队列：

• 等待 worker 的任务队列；
• worker 与 MCP 服务器之间的事件队列；
• 服务器端流式库内部的缓冲。

为每条队列设定合理的深度上限非常重要。若队列溢出，要么直接回应客户端“系统繁忙，请稍后重试”，要么丢弃低重要性的 job，要么把部分场景切到“离线模式”（例如生成报告后再发链接）。

还有一个有趣的做法是事件类型优先级化。在过载时，只发送 job.completed 与 job.failed，而把 job.progress 降级或暂时关闭。

4. 流与事件的监控

没有度量，rate‑limits 与 backpressure 就沦为“巫术”。你需要看见：流数量异常增多、事件开始滞后、客户端大量断开。

流与普通 HTTP 请求表现不一样：它们可能持续数分钟到数小时，因此“每秒请求数”“平均延迟”这类经典指标并不能给出全貌。

关键指标

对 SSE 或 HTTP/stream，建议关注以下几组指标。

1. 连接指标。 当前有多少条活动 SSE 流？单条连接平均寿命多长？多少比例的流以错误或超时结束？活动连接数的骤增提示潜在的流量风暴或资源泄漏（客户端未主动关闭）；骤降则提示大面积中断（例如网络问题或服务器的致命缺陷）。
2. 事件指标。 全部流每秒发送多少事件（EPS —— events per second，每秒事件数）？事件的平均大小？观察到多少反序列化或 payload 校验错误？如果事件体积突然增大，可能有人把本应是简短字符串的 job.progress 换成了整段报告文本。
3. job 指标。 各状态分布（pending、running、completed、failed、canceled）、按任务类型的平均耗时、进入重试或死信（dead‑letter）的比例。这帮助定位问题是否在 worker 侧：外部 API 变慢了，或出现了大规模错误。
4. backpressure 与系统层面的指标。 观察组件间缓冲与队列的深度，以及流因等待消费者“腾位置”而被阻塞的时间比例。如果队列几乎常年满仓，这是系统已到极限的明确信号。同时关注负责流传输的服务器 CPU/内存与网络层的错误/超时。有时 MCP 服务器与 ChatGPT 之间的网络带宽就是瓶颈。

综合这四组，你能回答三件事：当前有多少流在存活、你在传多少数据、job 的表现如何、系统具体在哪里“憋住了气”。

记录什么日志

日志是可观测性的第二支柱。要以便于回溯到某个特定 job 的方式记录事件与连接。

通常为每个事件与流在日志中附加：

• jobId 与/或 eventId；
• userId 与 sessionId（若有多租户）；
• 事件类型（progress、completed、failed、resource.updated）；
• 通道类型（SSE 或 HTTP/stream）；
• 发送时间戳，以及尽可能记录 worker 生成事件的时间戳。

这样可以计算 lag：worker 生成事件时间与其被写入 socket 的时间差。该 lag 的上升是 backpressure 问题的良好指示器。

注意别让日志本身成为过载来源。对高频事件如 job.progress，未必要记录每一条；可以做采样（只记录每第 N 条），或改为记录聚合统计。

代码层面可以是一个简单的 helper：

function logEvent(event: {
  type: string;
  jobId: string;
  userId?: string;
  channel: "sse" | "http-stream";
  payload: unknown;
}) {
  console.info({
    ...event,
    timestamp: new Date().toISOString(),
  });
}

在真实项目中，可以把它封装进结构化日志库，思路相同：在每条记录里尽量包含有用上下文。

5. 告警与降级策略

有了指标与日志，下一步就是配置告警，并设计系统在“生病”时如何“优雅地变差”。理念是：坦诚地“差一点”总比突然崩溃好。

告警示例

对 GiftGenius，值得关注几类常见情形。

首先，活动流数量异常。如果平时只有几十条 SSE 活动连接，突然变成上千条，需要查明原因。可能你变得很受欢迎，也可能是缺陷导致连接未被关闭。

其次，job 实际完成与客户端收到 job.completed 的延迟。如果该延迟超过阈值（比如 5–10 秒），说明 worker 与客户端之间的某处在积压事件或连接阻滞。

第三，job.failed 或 job.canceled 的占比居高不下。原因可能在 worker（外部 API 挂了、新缺陷），也可能在用户对延迟变得更敏感（更频繁地取消任务）。

最后，连接错误与流中断升高：若异常断开（disconnect）数量增加，可能是网络或客户端的问题，值得考虑 fallback 场景。

降级模式（Degradation patterns）

系统过载时，可以启用“资源省用”模式。这总比一股脑儿 500 更好。

最常见的模式是自适应事件频率。如果看到 event‑rate（每秒事件数）突然飙升到常态的十倍，且队列里的滞后在增长，就降低进度事件频率。原来每 1% 一次，改为每 10%；原来每 500 ms 一次，改为每 2–3 秒。用户完全能接受少些“精细进度”，但无法接受完全卡死的 UI。

对于不太重要的事件——例如商品 feed 的后台 resource.updated——可以在系统过载期间暂时停发。

另一招是把部分场景从流切换到周期性轮询（polling）。如果 SSE 通道在崩，MCP 服务器可以向小部件发送类似 system.overloaded 的系统事件，小部件则切换为“每隔 N 秒轮询 REST endpoint 以获取 job 状态”的策略。

6. GiftGenius 的一个小型实践片段

把以上串起来，假设我们已有：

• MCP‑tool startGiftSearch，用于创建 job 并返回 jobId；
• 执行搜索的 worker，它会发送 event/progress 与 event/completed；
• 小部件在 Next.js 上连接的 SSE endpoint：/api/events/[userId]。

加上一层简单的“防事件风暴”保护与最小化的监控。

按步长与时间限制进度

在 worker 中加入节流与合并，如上所述。现在事件不超过每半秒一次，且变化至少 5% 才发送。

记录活动流

在 SSE endpoint 按用户维护计数器：

const activeStreams = new Map<string, number>();
const STREAM_LIMIT = 3;

function canOpenMoreStreams(userId: string) {
  const current = activeStreams.get(userId) ?? 0;
  return current < STREAM_LIMIT;
}

function registerSseClient(userId: string, controller: ReadableStreamDefaultController) {
  const current = activeStreams.get(userId) ?? 0;
  activeStreams.set(userId, current + 1);

  // 在这里把 controller 保存到某个结构中，
  // 以便随后向该流写入事件
}

function unregisterSseClient(userId: string) {
  const current = activeStreams.get(userId) ?? 1;
  activeStreams.set(userId, Math.max(0, current - 1));
}

服务器还可以把关于 activeStreams.size 的指标上报到 Prometheus/Grafana 或任意监控系统。

最简单的 event‑rate 指标

先粗略统计一下我们发了多少事件：

let eventsSentLastMinute = 0;

function sendProgressToClient(ev: ProgressEvent) {
  // ... 序列化并写入 SSE 流
  eventsSentLastMinute++;
}

setInterval(() => {
  console.info({
    metric: "events_per_minute",
    value: eventsSentLastMinute,
    timestamp: new Date().toISOString(),
  });
  eventsSentLastMinute = 0;
}, 60_000);

后续可以替换为正规的计数器与告警，但作为起点已经不错。

把上述拼起来：限流、backpressure、指标/告警与合理的 UX fallback，能让你的 GiftGenius 不再是“只适合做演示的 demo”，而是能扛住真实的流量风暴。接下来的模块里，我们会讨论 gateway、生产架构与完善的可观测性，这些模式还会持续发挥作用。

7. 处理流、rate‑limits 与监控时的常见错误

错误 1：没有对流数与事件频率做限制。
开发者为了“更炫”而加入了 SSE，worker 老老实实地在每个对象处理完就发一次进度，在 demo 上一切如常。但第一次遭遇真实用户高峰时，服务器的大部分资源都被用在了序列化与传输成千上万条碎片事件上，而 ChatGPT 里的 UI 变成了幻灯片。

错误 2：试图“无限缓冲一切”。
代码里出现了无限增长的“未发送事件数组”，它会一直长到客户端“恢复”为止。剧透：客户端未必会恢复，服务器先死。任何缓冲都应有硬上限，溢出处理逻辑必须显式。

错误 3：把所有事件当成一类对待。
进度可以聚合并丢弃（最后的百分比比历史轨迹更重要）。日志与 partial 结果不行——丢一个分片就可能破坏数据。设计系统时，请预先按重要性对事件分组，并为每组设计过载策略。

错误 4：缺乏可观测性。
没有活动流的指标、没有 event‑rate 统计，日志里只有“出错了”。在这种情况下，你只能从用户反馈与 CPU 负载曲线里得知问题。至少按 jobId 与 eventId 配置基础指标与日志，并非奢侈，而是刚需。

错误 5：UX 刚性，不考虑降级。
小部件与 GPT 指令默认流总是可用、进度“实时”更新、partial 结果严格按剧本到达。一旦遇到网络问题，用户就看到“挂住”的进度条与零解释。更好的做法是提供诚实的 fallback：“实时更新出了点问题，但我会继续为你挑选并在完成时告知”，同时切换到更低频的更新或轮询。

错误 6：相信“我们的用户不会创建很多并发任务”。
实践一再表明，如果你不限制并行 job 与流的数量，总会有人开五个标签页、在每个页里把选礼“拉满”，然后去喝咖啡。“也许不会出事”的想法在生产上几乎总会以告警声大作、你开始学习监控为结局。