幕 IV · 深水区CH 21进度 21/25

背压：为什么你的「异步」会 OOM

同步阻塞有一个被严重低估的优点：它天然限流。你只有 200 个线程，所以最多只能同时处理 200 个请求 —— 第 201 个请求会被迫等待。这个「等待」很难看，但它是一道保险丝。而异步把这道保险丝拆了。现在你可以「先接下来再说」—— 于是你接下了永远处理不完的活，它们排在一个队列里，静静地涨，直到内存耗尽。这一章讲的是异步的账单：无界队列不是「没有背压」，它只是「把背压推迟到了 OOM 那一刻」。

Little 定律无界队列四种策略限流雪崩

一个真实的事故

// 一个「性能优化」:把同步的日志上报改成异步的
class Reporter {
    private val scope = CoroutineScope(Dispatchers.IO)

    fun report(event: Event) {
        scope.launch {              // ← 不阻塞调用方了!性能提升!
            httpClient.post(url, event)   // 200ms
        }
    }
}

// 上线。压测。QPS 从 800 涨到了 3000。
// 大家很开心。
//
// 两周后,一次大促,流量涨到 5000 QPS。
// 服务在 40 分钟后 OOM。所有实例,几乎同时。
这 40 分钟里发生了什么λ（到达）= 5000/s，μ（处理）= 5000/s？不，只有 320/s
上报请求 每秒来 5000 个，每个 launch 一个协程
HTTP 客户端 64 并发 连接池就这么大 —— 实际处理能力 ≈ 64/0.2s = 320/s
堆积中的协程 每秒净增 5000 − 320 = 4680 个。每个协程还引用着那个 Event 对象。
算账 40 分钟 = 2400 秒 × 4680 = 1123 万个协程，每个还挂着一个 Event 对象。
OOM 是必然的，只是时间问题。
而在 OOM 之前，最后一个上报的事件，已经排了 9 个小时的队。
◆ 定论 · scope.launch { } 是一个隐形的无界队列

你以为你在「异步化」。你实际上是在建一个没有上限的待办清单。

同样的犯罪现场，换个语言换个面孔，但本质完全一样：

  • Kotlin：scope.launch { } 在一个循环里 / Channel(Channel.UNLIMITED)
  • Go：go f() 在一个循环里 / make(chan T, 100000)
  • Python：asyncio.create_task() 在一个循环里 / asyncio.Queue() 不带 maxsize
  • Java：Executors.newCachedThreadPool() / newFixedThreadPool 配默认的无界 LinkedBlockingQueue

它们全都是同一个 bug 的不同拼写。

Little 定律：在 OOM 之前，延迟已经先崩了

排队论里最简单也最有用的一条定律（John Little, 1961）：

L = λ × W

  L = 系统里「同时在飞」的请求数（队列长度 + 正在处理的）
  λ = 到达率（QPS）
  W = 每个请求在系统里待的总时间（排队 + 处理）

反过来:
  W = L / λ        ← 队列越长,延迟越高。线性关系。
◆ 定论 · 队列长度就是延迟

假设你的服务能处理 1000 QPS（μ = 1000），队列里排了 1 万个请求（L = 10000）。

那么新来的那个请求，要等 10000 / 1000 = 10 秒才被处理。

而你的客户端超时是 3 秒。

所以：这 10000 个请求里，绝大多数在被处理的时候，客户端早就超时走人了。你的服务器在100% 满负荷地做无用功 —— 它在处理一堆没有人还在等的请求。

这就是「队列积压」最恶毒的地方：它不只是慢，它会让你的系统进入一个「所有工作都白做」的稳态。而且它自己爬不出来 —— 因为处理这些废请求要占用资源，导致新请求排得更后面，导致更多请求超时……

这叫「队列崩溃」（queue collapse），或者更通俗的名字：雪崩。

四种策略

下面这个模拟器把 λ 和 μ 拿出来给你拖。先把 λ 调到大于 μ（这是所有事故的前提），然后逐个切换四种策略。

一张选择表

策略内存延迟丢数据吗什么时候用
① 无界队列 无界 → OOM 无界 → 雪崩 不丢（但你会 OOM，然后全丢） 永远不要。
它唯一的用途是「我确定 λ 永远小于 μ」—— 而你不能确定。
② 有界 + 阻塞
（真背压）		 有界 ✓ 有界 ✓ 不丢 默认选它。
压力沿管道往回传，一路传到源头（TCP 窗口 → 客户端发得慢 → 或者你返回 429）。
系统会变慢，但不会死。
③ 有界 + 丢弃 有界 ✓ 有界 ✓ 丢 数据本来就可丢时：监控打点、日志采样、鼠标移动事件、实时视频帧。
订单、支付、任何「恰好一次」的语义：绝对不行。
④ conflate
（只留最新）		 O(1) ✓ 最低 ✓ 丢中间值 UI 状态的正解。
ViewModel 每秒推 120 个状态，屏幕只有 60Hz —— 中间那些状态渲染出来也没人看得见。
Kotlin：StateFlow 天生 conflate。

三门语言怎么写「有界 + 阻塞」

// ✅ 用有界 Channel(满了 send 就挂起 —— 这就是背压)
private val queue = Channel<Event>(capacity = 256)

fun report(event: Event) {
    // trySend:非阻塞。满了返回失败,你可以选择丢弃或者记指标
    val ok = queue.trySend(event).isSuccess
    if (!ok) droppedCounter.increment()      // ← 至少你【知道】自己丢了
}

// 或者,如果调用方是 suspend 的,直接背压给它:
suspend fun report(event: Event) {
    queue.send(event)     // ← 满了就挂起。调用方被顶住了。这才是背压。
}

// 固定数量的 worker 消费:
init {
    repeat(8) {
        scope.launch {
            for (event in queue) {   // ← 只有 8 个协程在跑,不会爆
                httpClient.post(url, event)
            }
        }
    }
}

// ✅ Flow 的版本(更地道)
events
    .buffer(256)              // 有界缓冲。满了 → 上游 emit 挂起 → 背压
    .flowOn(Dispatchers.IO)
    .collect { post(it) }
// ✅ 有界 channel + 固定 worker 数
type Reporter struct {
    ch chan Event
}

func NewReporter() *Reporter {
    r := &Reporter{ch: make(chan Event, 256)}   // ← 有界!
    for i := 0; i < 8; i++ {                    // ← 只有 8 个 worker
        go r.worker()
    }
    return r
}

func (r *Reporter) Report(e Event) {
    select {
    case r.ch <- e:              // 有空位 → 塞进去
    default:                     // 满了 → 【不阻塞,直接丢】+ 记指标
        droppedCounter.Inc()
    }
}
// 想要真背压(阻塞调用方)?去掉 default:
//   r.ch <- e    ← 满了就阻塞在这里。调用方被顶住。

func (r *Reporter) worker() {
    for e := range r.ch {
        post(e)
    }
}
import asyncio

queue = asyncio.Queue(maxsize=256)      # ← 【一定要设 maxsize】

async def report(event):
    await queue.put(event)               # ← 满了就挂起。这就是背压。
    # 或者不想阻塞:
    # try:
    #     queue.put_nowait(event)
    # except asyncio.QueueFull:
    #     dropped_counter.inc()

async def worker():
    while True:
        event = await queue.get()
        try:
            await post(event)
        finally:
            queue.task_done()

# 启动固定数量的 worker
async def main():
    async with asyncio.TaskGroup() as tg:
        for _ in range(8):
            tg.create_task(worker())
◆ 定论 · 三条通用规则
  1. 每一个队列都必须有界。没有例外。「这个队列不可能积压」这句话，历史上说过的人都错了。
  2. 消费者的数量必须是固定的、有上限的。不要「来一个任务开一个协程」—— 那等于没有限流。
  3. 队列满了之后的行为，必须是一个显式的决定。阻塞？丢弃？返回 429？—— 三个都可以，但你必须明确选一个，并且给它打上指标。
    最糟的情况不是「丢了数据」，是「丢了数据而你不知道」。

背压要传到源头，才算真的背压

光在自己这一层加个有界队列，只是局部的背压。真正的背压要一路传回到最开始的那个人：

背压的完整传播链压力从右往左传
客户端 收到 429 → 退避重试（或者被 TCP 窗口顶住，发不动了）
你的服务 处理中队列满 → 拒绝新请求 / 不再从 socket 读
你的队列 256/256 满了
下游（DB） 它就这么快，改变不了
关键 最左边那个箭头是最重要的。如果压力传不到客户端，客户端就会继续以原速度发 —— 那你的队列迟早还是会爆，只是爆在别的地方。
「返回 429 / 503」不是认输，是唯一负责任的做法。它告诉客户端：「慢一点，不然我们一起死。」
◈ 三语 · 背压是怎么被 TCP 白送给你的

一个很多人不知道的事实：如果你写的是同步阻塞的服务，TCP 已经免费给了你背压。

  • 你的线程忙不过来 → 没人从 socket 读 → 内核的接收缓冲区满了
  • → TCP 的滑动窗口变成 0（「零窗口通告」）
  • → 对端的 TCP 栈停止发送
  • → 对端应用的 write() 阻塞了
  • → 压力一路传回到了源头。而这一切，你一行代码都没写。

而当你「异步化」之后，你亲手把这个机制拆了 —— 因为你现在会飞快地把 socket 读空（不读空事件循环就不转），然后把数据扔进一个内存队列。TCP 再也不会帮你踩刹车了。

所以：异步系统必须显式地重建背压。这不是一个可选的优化 —— 这是你为了拿到异步的性能，必须偿还的一笔债。

还有一件事：超时不等于取消（预告）

// 队列积压时,这个「优化」看起来很聪明:
ctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
defer cancel()
result, err := doWork(ctx)   // 超过 100ms 就放弃

// ❌ 但如果 doWork 不检查 ctx —— 那个活儿【还在跑】。
//    你「不等了」,但下游的负载一点没减。
//
// 在雪崩的场景里,这是致命的:
//   你以为你在快速失败,实际上你在【以最快的速度】往下游堆积无用的工作。

这是下一章的主题。

✚ 动手

1. 全局搜你项目里所有的队列/缓冲/线程池，检查它们是不是有界的。
关键词：newCachedThreadPool、LinkedBlockingQueue()（无参 = 无界！）、Channel(UNLIMITED)、asyncio.Queue()（无 maxsize）、make(chan T, 100000)（这也是「事实上的无界」）。

2. 搜「在循环里启动并发任务」的地方（for { go f() }、for { scope.launch {} }、for { create_task() }）。那些全都是隐形的无界队列。

3. 用 Little 定律算一次你的服务：QPS × 平均延迟 = 同时在飞的请求数。这个数和你的线程数/并发上限对得上吗？

4. 给你的队列加一个「当前长度」的监控指标。如果这条曲线在单调上升 —— 你已经在事故里了，只是还没爆。

下一章：取消。它是并发里最难的那部分 —— 因为没有人能强行杀死一个正在跑的任务，只能请它自己停。