Go NATS Go 订阅的 pending 限额为什么影响慢消费者

Go 使用 github.com/nats-io/nats.go 做异步订阅时，回调处理速度一旦低于消息到达速度，订阅内部就会积累 pending 消息。pending 限额不是“让回调变快”的开关，而是客户端为积压设置的两道边界：消息数量上限和字节数上限。任意一道先达到，订阅就可能进入慢消费者状态并丢弃后续消息。

排查这类问题时，先看 Pending() 和 Dropped()，再看当前 PendingLimits()；不要一上来把限额改成无限。下面用一个架构取舍来解释：限额保护的是哪段内存，为什么会影响慢消费者，以及怎样把它变成可观察、可控制的配置。

pending 到底限制了哪一段队列

对 core NATS 异步订阅来说，服务端把消息送到客户端后，nats.go 需要先把消息放进对应订阅的内部队列，再交给 MsgHandler。如果回调里有慢查询、磁盘写入或外部 HTTP 调用，队列就会比消费端更快地增长。Pending() 返回的正是这个客户端订阅当前排队的消息数和字节数。

因此，pending 不等于 JetStream 消费者的 MaxAckPending。前者是 nats.go 客户端订阅侧的排队边界，后者属于 JetStream 消费者的服务端确认控制。两者都可能让“消息没有继续交给处理逻辑”，但所在位置和调整方法不同。

两个限额谁先触顶，取决于消息形状

nats.go 当前文档给出的默认订阅 pending 限额是 500,000 条消息和 64 MiB。它们是两个独立维度：小消息密集时，消息数可能先到；单条消息较大或带有较多 header 时，字节数可能先到。字节统计包含消息数据和编码后的 header，因此只按条数估算内存并不稳妥。

这也是为什么“把消息数上限调大”不一定解决问题：如果真正受限的是字节数，条数上限改大只会让判断更晚出现。反过来，单纯提高字节上限也可能让大量小消息在内存中堆积。

用 nats.go 的接口把慢消费者信号接出来

先给订阅设置异步错误回调，再在定时观测中读取当前积压、历史峰值和丢弃量。示例只展示观测与配置关系，不要求在本机启动 NATS 服务；回调中的业务处理应替换成自己的逻辑。

package main

import (
    "errors"
    "log"
    "time"

    nats "github.com/nats-io/nats.go"
)

func main() {
    // 把慢消费者错误接到统一处理入口，便于记录 subject 和限额变化。
    nc, err := nats.Connect(nats.DefaultURL,
        nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
            if errors.Is(err, nats.ErrSlowConsumer) {
                log.Printf("slow consumer: %v", err)
            }
        }),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Close()

    // 这里设置有界缓冲；两个维度都要结合单条消息大小来估算。
    sub, err := nc.Subscribe("orders.created", func(msg *nats.Msg) {
        // 真实业务中应避免在回调里长时间阻塞消息分发。
        _ = msg
    })
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Unsubscribe()

    if err := sub.SetPendingLimits(100_000, 32*1024*1024); err != nil {
        log.Fatal(err)
    }

    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()
    for range ticker.C {
        // Pending 是当前积压，MaxPending 是曾达到的峰值，Dropped 是已知丢弃量。
        pendingMsgs, pendingBytes, err := sub.Pending()
        if err != nil {
            log.Printf("read pending: %v", err)
            continue
        }
        maxMsgs, maxBytes, _ := sub.MaxPending()
        dropped, _ := sub.Dropped()
        log.Printf("pending=%d/%d bytes=%d/%d max=%d/%d dropped=%d",
            pendingMsgs, 100_000, pendingBytes, 32*1024*1024,
            maxMsgs, maxBytes, dropped)
    }
}

Pending() 是当前值，MaxPending() 是订阅生命周期内观察到的最大排队量，Dropped() 统计的是客户端已知因 pending 限额而丢弃的消息。官方文档特别提醒：如果是服务端把连接判定为 Slow Consumer，Dropped() 的数值可能不完整，所以它应与错误回调和连接指标一起看。

SetPendingLimits 的三个边界决定了取舍

设置限额时，最容易忽略的是参数语义：

• 第一个参数限制排队消息数，第二个参数限制排队字节数。
• 0 不允许；传入 0 会返回错误。
• 负数表示对应维度不限制，两个参数都设为负数就相当于不设客户端订阅上限。

// 只放宽消息数，仍用字节数保护内存。
if err := sub.SetPendingLimits(200_000, 64*1024*1024); err != nil {
    // 配置错误应在启动阶段暴露，而不是运行中静默忽略。
    return err
}

// 读取最终生效的两个维度，避免只相信配置文件中的期望值。
msgLimit, byteLimit, err := sub.PendingLimits()
if err != nil {
    return err
}
log.Printf("pending limits: messages=%d bytes=%d", msgLimit, byteLimit)

“无限制”只解决了客户端尽早丢弃的问题，却把内存风险交给了进程。慢回调没有改善时，队列仍会继续增长；在容器或多租户进程中，这种配置可能把一个订阅问题放大成整进程的内存压力。因此限额应被视为背压边界，而不是性能优化开关。

客户端慢消费者和服务端慢消费者不是一回事

客户端订阅慢消费者通常是：连接读线程已经拿到消息，但消息在某个订阅的 pending 队列中排队，回调消费不过来。服务端连接级 Slow Consumer 则是服务端向某个连接写数据时，连接上的待发送数据达到服务端限制；NATS 协议文档把默认连接待发送上限描述为 10 MB。两种情况都可能看到“慢消费者”，但诊断位置不同。

排查时可以按边界问三个问题：消息是否已经抵达客户端连接、pending 是否持续接近某个限额、错误来自订阅异步错误回调还是连接关闭原因。只有前两个答案都指向订阅队列时，才应该优先调整 SetPendingLimits；如果连接本身读不动或服务端写不出，扩大订阅队列并不能修复根因。

把 pending 限额变成可执行的判断清单

可以把处理顺序固定成下面这份清单：

1. 记录 Pending() 的消息数和字节数，确认是哪一维接近上限。
2. 读取 PendingLimits()，确认实际生效的值，避免把默认值或旧配置当成事实。
3. 同时记录 MaxPending() 与 Dropped()，把瞬时积压和历史丢弃分开。
4. 检查回调是否在做同步 I/O、锁等待或过重的反序列化；优先缩短回调临界路径。
5. 如果业务允许，使用受控的 worker 分流，但要把并发度、任务队列和关闭流程一并纳入内存预算。
6. 最后再提高单订阅限额，并为消息数和字节数分别设上界。

判断标准很简单：限额提高后，Dropped() 不再增长并不代表系统已经健康，只能说明“更晚丢”。只有回调吞吐、积压峰值、内存占用和错误信号一起回到可接受范围，慢消费者问题才算真正收敛。

常见追问

pending 限额调大后为什么还是会丢消息？

可能是字节数维度先触顶，也可能已经进入服务端连接级 Slow Consumer；还可能是回调处理速度长期低于入口速率。先分别读取两个 pending 值和限额，再看连接级错误。

能不能把两个 pending 限额都设成 -1？

API 语义允许负数表示不限制，但这会取消客户端订阅队列的内存边界。只有在消息速率、单条大小和进程内存都有明确上限时，才有理由采用这种策略。

为什么只看 Dropped 不够？

Dropped() 反映的是客户端已知因 pending 限额丢弃的数量；官方文档说明，连接被服务端判定为 Slow Consumer 时这个数可能不完整，所以还要结合异步错误和连接状态。

总结

Go NATS Go 订阅的 pending 限额之所以影响慢消费者，是因为它直接定义了客户端订阅队列能承受多少条消息、多少字节。消息数和字节数任一道先触顶，都可能带来 ErrSlowConsumer 与丢弃。最稳妥的做法是先用 Pending、PendingLimits、MaxPending 和 Dropped 建立证据，再优化回调和分流，最后才在内存预算内放大有界缓冲。

官方资料：https://pkg.go.dev/github.com/nats-io/nats.go