Go NATS Go 客户端怎样让多个订阅者分担同一主题消息

我第一次把一个 Go NATS 消费者复制成三个进程时，发现同一条订单消息被三个进程都打印了一遍。连接是通的，主题也没有写错，但这不是“自动分担”，而是普通发布订阅的正常行为：同一主题上的普通订阅者都会拿到自己的副本。

如果目标是让多个 worker 共同处理一个主题，应使用 nats.go 的 QueueSubscribe，并让这些订阅使用完全相同的队列组名。官方入口：

官方地址：https://pkg.go.dev/github.com/nats-io/nats.go

它解决的是组内分摊，不是消息持久化。Core NATS 属于即时投递模型，消费者在收到后崩溃，不能把它当成带重试的任务队列。

先看普通订阅为什么会重复处理

普通的 Subscribe 只表达“我对这个主题感兴趣”。同一个主题有三个普通订阅者时，发布一条消息，三个订阅者都会收到一份。这适合通知、审计或指标采集，因为每个观察者都需要看到完整事件流。

// 普通订阅：每个实例都会收到 orders.created 的消息。
nc.Subscribe("orders.created", func(msg *nats.Msg) {
    // 这里代表通知、审计等需要各自保留消息的处理方。
    fmt.Println("received:", string(msg.Data))
})

所以，单纯启动多个副本不会形成 worker 池。要分担工作，关键不是连接数，而是订阅时加入同一个队列组。

用 QueueSubscribe 让多个 worker 共享消息

nats.go 的 QueueSubscribe(subject, queue, callback) 会创建异步队列订阅。多个订阅者只要主题相同、第二个参数也相同，就属于同一个 queue group；每条匹配消息只会在这个组里选择一个成员。

package main

import (
    "fmt"
    "log"

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

func main() {
    nc, err := nats.Connect(nats.DefaultURL)
    if err != nil {
        log.Fatal(err)
    }
    // 退出时优雅排空连接，避免已经交给客户端的消息被突然截断。
    defer nc.Drain()

    for _, workerName := range []string{"worker-a", "worker-b", "worker-c"} {
        name := workerName
        _, err := nc.QueueSubscribe("orders.created", "order-workers", func(msg *nats.Msg) {
            // 三个实例使用同一个组名，因此一条消息只会进入其中一个回调。
            fmt.Printf("%s handles %s\\n", name, string(msg.Data))
        })
        if err != nil {
            // 订阅失败时停止启动，避免形成只有部分 worker 的假象。
            log.Fatal(err)
        }
    }

    // 确认订阅请求已经发给服务器，再进入长期消费状态。
    if err := nc.Flush(); err != nil {
        log.Fatal(err)
    }
    select {}
}

这里真正起作用的是 "order-workers"。三个 worker 可以来自三个进程、三个容器，甚至三台机器，只要它们连接到同一个 NATS 集群，并在相同主题上使用字节级一致的组名，服务端就把它们视为一个逻辑订阅者。

排查“还是重复消费”时先对齐这三个字段

我排查这类问题时，先把每个实例的订阅参数打印到启动日志，再逐项比对，而不是先怀疑 NATS 没有负载均衡。

组名是精确匹配的。order-workers 和 order-worker 看起来只差一个字母，但服务端会把它们当作两个组；于是每条消息会分别交给两个组各自的一个成员，结果仍然像“重复处理”。

普通订阅和队列组也可以共存。例如审计服务用 Subscribe 接收每一条订单，订单处理池用 QueueSubscribe 分担处理，它们不会互相抢消息：前者每条都收，后者每组只选一个。

不要把 Core NATS 队列组当成可靠任务队列

队列组解决的是“同一时刻由谁处理”，并不负责“处理失败后是否找回”。Core NATS 是 at-most-once 语义：服务端把消息交给某个组成员后，如果该成员在处理过程中退出，消息不会自动转给另一个成员。

因此可以按下面的边界选择：

• 通知、实时刷新、短生命周期的工作分担：使用 Subscribe 或 QueueSubscribe。
• 要求持久化、确认、重投、消费进度和故障恢复：改用 NATS JetStream 的持久化消费者。
• 需要严格顺序或业务幂等：在消费者侧设计幂等键，不要把“同一组只选一个成员”误当成业务去重。

还有一个容易忽略的现象：组内分配不应被假设为严格轮询。某个 worker 连续拿到两条消息并不代表组配置失效；只要组名相同，服务端仍是在组成员中选择一个可用成员。扩缩容时，新成员加入后会参与后续分配，退出成员则不再接收新消息。

最后用一张清单确认配置

如果你的目标是让多个 Go 实例分担同一主题，最小条件就是：连接到同一个 NATS 服务；订阅完全相同的 subject；都调用 QueueSubscribe；queue 参数逐字一致；业务处理具备幂等性。若任务不能丢，再把可靠性要求交给 JetStream，而不是继续增加普通订阅者。

相关问题：

• 为什么多个 Go NATS 普通订阅者会同时收到消息？因为普通订阅是发布订阅语义，每个匹配者都会收到一份。
• 队列组名称能不能不同？不能。不同名称代表不同组，会各自获得一份组内消息。
• QueueSubscribe 能保证失败重试吗？Core NATS 不能；需要持久化与重投时应使用 JetStream。