背压处理：给AI流式接口装上减压阀

可视化演示

三步展示背压处理：速率不匹配场景（prod快→buf溢出→cons慢）、环形缓冲区水位管理（low/high/full三级）、令牌桶速率适配（rate R + capacity C + consume控制）。

背压原理：水满了要会喊停

你是一栋写字楼的物业主管。楼顶水塔负责向每间办公室供水，它想多快就多快；可住户的水杯就那么大，喝得慢。水塔猛灌，水管里的水越积越多，最后从杯口溢出，甚至接头爆开。流式 API 的世界里，水塔就是大语言模型推理服务，水杯就是客户端，而管道里流动的是一块块正在生成的文字。背压处理，就是在这条管线上建立的“喊停”机制。

当模型每秒生成 20 个 token，而客户端渲染每秒只能消化 5 个 token，生产速度和消费速度之间就出现了差值。这个差值如果不加以干预，会变成内存里的消息积压、网络缓冲区的膨胀，最后表现为连接超时、消息丢失，用户看到一句话说到一半就断了。背压信号就是“下游快满了，请上游慢一点”的反馈，它可以沿数据链路一层层向上传递，直到源头调整速度。

背压的本质是让上游能感受到下游的“杯子满了”。在后续设计中，蓄水池水位线、智能阀门和双向对讲，都是围绕这个核心展开。

SSE背压：单向水管上的自动阀门

现在改用一根单向水管：水塔只管把水送进办公室，住户没法顺着水管喊回去。SSE（Server-Sent Events，服务器推送事件）就是这样一根单向管——服务端持续向客户端推送事件，客户端只能接收，不能反向给服务端发消息。它天然适合 AI 聊天框里“一句话一句话往外蹦”的效果。

既然不能喊话，客户端如何让水塔慢下来？答案是调节进水口的阀门。浏览器把 SSE 流当成一个可暂停读取的数据流（ReadableStream），你不需要一次性把所有数据读完，而是读完一段、处理完，再继续读下一段。当自己内存里的待渲染队列接近上限，就暂时不去调用读取接口；水塔那边发现管道堵了，TCP 层的窗口会自动收缩，相当于感受到了背压。

实践上还要注意：不能因为暂停读取就让连接被误判成死链，所以服务端要定期发心跳注释，保证中间网络不回收；客户端也要防止渲染队列无限增长，设置一个合理的水位阈值。这样，单向水管虽然没有对讲机，却依然能用“阀门开度”完成一场无声的谈判。

WebSocket背压：双向对讲机让谈判更高效

如果水管旁边再配一部对讲机，住户就能直接朝水塔喊：“慢点儿放！”WebSocket 就是这样的双向通道，客户端和服务端都可以主动发消息。它不仅适合服务端推送，也适合客户端上报自己的处理进度。对 AI 助手类应用来说，WebSocket 常用来承载多轮对话、工具调用事件和中间状态更新。

在 WebSocket 场景下，背压主要靠三个抓手。第一，发送方发送前先看发送缓冲区（bufferedAmount）有多大；如果本地已经积压了太多待发消息，就暂停发送。第二，接收方每处理完一条完整消息，回一个 ACK 确认；发送方维护一个“未确认窗口”，窗口满了就停下来等 ACK。第三，超出窗口的消息先留在业务队列中，而不是塞进网络层。

对讲机不是用来吵架的，而是用来同步节奏的。比起 SSE 的“阀门暗示”，WebSocket 可以让双方就“每批发几条、何时继续”达成明确协议，背压的执行效率更高，也更容易在分布式系统里做流控。

缓冲区管理：蓄水池要画水位线

无论单向还是双向，中间最好有一个蓄水池：它缓冲短时高峰，避免水直接冲击住户。缓冲区本质上是一个队列，生产方向队尾追加消息，消费方从队头取走消息。蓄水池不可能无限大，内存是硬约束；我们需要画两条水位线。

高水位线代表“快满了，请上游减速或暂停”，低水位线代表“已经缓过来了，可以继续供水”。两条线之间留一段滞回区间，防止生产者刚停又要启动，像水龙头反复开关一样磨损设备。

C 的取值来自内存预算：如果每条消息平均占 1 KB，内存上限 64 MB，C 最多设为 65536；实际还要留出系统余量。对于 LLM 流式输出，还可以在推理循环内部给生成器一个“节流信号”，让模型放缓生成，而不是把 token 大量堆进队列。蓄水池不是无限大的面子工程，水位线才是它的灵魂。

速率适配：智能减压阀平滑调速

一开一停虽然简单，但会让水流忽大忽小，管道里产生水锤。更优雅的做法是给水塔装一个智能减压阀：看到下游压力偏高时，不是直接关闸，而是按比例把出水量调小；看到下游跟得上，再慢慢开大。

这个思想落到流式 API 上就是反馈控制。以当前缓冲区积压量 $B_{now}$ 与目标水位 $B_{target}$ 的偏差作为反馈，生产速率 $r_{new}$ 在基准速率 $r_{base}$ 上做比例调节。

除了比例控制，工程中还常用令牌桶：系统以固定速率往桶中放令牌，每次发送要拿走一张令牌；桶里最多攒几张令牌，所以允许小规模突发，平均速率始终受控。这就相当于水塔每个月发固定数量的水票，偶尔可以预支，但不会无限透支。有了智能阀门，管道不用反复经历“急停-猛启”的折磨。

断开恢复：从上次断口接着接水管

工地上的水管常因车辆碾压而破损。抢修时，维修工人不会重新铺一整条新管，而是找到上次接头的断口，把新管接到旧口上。流式连接也是如此——一次 LLM 推理可能长达几十秒，任何一次网络抖动都可能让连接中断；如果每次都让模型从头生成，用户体验是一场灾难。

解决方案是给消息编号。服务端为每一条流式消息分配一个单调递增的序号（offset），客户端在处理成功后记录最后一条已处理序号 $N_{last\_acked}$；重连时把这个序号带给服务端，服务端从下一条开始继续推送。

客户端还要配合幂等去重——同一序号如果重复到达，直接忽略。服务端需要缓存最近 N 条消息用于补发，等客户端 ACK 之后慢慢释放缓存。记录断口，就是给断线修复留一张施工图。

方案设计与验证：整栋楼的给水工程验收

把前面所有部件组合起来，就是一张整栋楼的给水工程图：模型推理服务是水塔，发送队列是蓄水池，SSE/WebSocket 是水管，客户端是住户，水位线、智能阀门和序号续传是阀门与施工规范。

完整方案包括：在缓冲层设置高低水位；在发送层启动速率适配；在连接层实现断线重连与续传。上线前要做“通水测试”：人为把客户端消费速率调低，观察 buffer_usage 是否打满、lost 是否为零、平均端到端延迟是否可控。监控指标至少包括 producer_rate、consumer_rate、buffer_usage、lost、retry_count、p95_latency。

压测时对比无背压与有背压两条曲线：无背压时 buffer_usage 会撞到 100%，lost 爆发；有背压时 buffer_usage 稳定在高低水位之间，lost 归零。典型结果里，内存峰值下降 50% 到 70%，消息丢失从“偶发”变成“几乎为零”。验收标准不是水塔多能灌，而是住户杯子永远不溢出。

AI 应用场景

Python 演示

import time class BackpressureDemo:    def __init__(self, capacity=8, high=6, low=2, prod_rate=4.0, cons_rate=1.0):        self.capacity = capacity        self.high = high        self.low = low        self.prod_rate = prod_rate        self.cons_rate = cons_rate        self.buffer = []        self.produced = 0        self.consumed = 0        self.lost = 0        self.paused = False        self.enabled = False     def produce(self):        if self.enabled and len(self.buffer) >= self.high:            self.paused = True            return        if len(self.buffer) >= self.capacity:            self.lost += 1            return        self.buffer.append("token")        self.produced += 1     def consume(self):        if self.buffer:            self.buffer.pop(0)            self.consumed += 1        if len(self.buffer) <= self.low:            self.paused = False     def run(self, seconds=5):        start = time.time()        next_prod = start        next_cons = start        while time.time() - start < seconds:            now = time.time()            self.enabled = now - start >= 2.5            if not self.paused and now >= next_prod:                self.produce()                next_prod = now + 1.0 / self.prod_rate            if now >= next_cons:                self.consume()                next_cons = now + 1.0 / self.cons_rate            print("backpressure=%d buffer=%d produced=%d consumed=%d lost=%d paused=%s" % (                self.enabled, len(self.buffer), self.produced, self.consumed, self.lost, self.paused))            time.sleep(0.1) if __name__ == "__main__":    BackpressureDemo().run(seconds=5)

1. self.enabled = now - start >= 2.5前半段模拟无背压，后半段开启背压机制。
2. if self.enabled and len(self.buffer) >= self.high:高水位达到时暂停生产。
3. if len(self.buffer) >= self.capacity:缓冲区已满却继续生产，则消息丢失。
4. self.buffer.append("token")生产者把 token 放入队列尾部。
5. self.buffer.pop(0)消费者从队头取走一条 token。
6. if len(self.buffer) <= self.low:低水位时恢复生产。
7. next_prod = now + 1.0 / self.prod_rate按照生产速率安排下一次生产时间点。

常见误区

面试题