学完这篇，你能在 Go 项目里分别接上 NATS、RabbitMQ、Kafka 三套客户端，知道发消息、收消息、确认（ack）该怎么写，以及三者最容易踩的坑在哪。
第一步：先把选型和本地环境定下来
三个 MQ 定位差别很大，先别纠结：
- NATS：轻、快、延迟低，适合服务间通知、请求-响应。核心模式是 at-most-once，要持久化得开 JetStream。
- RabbitMQ：路由最灵活（exchange + routing key + binding），适合任务队列、工作分发。
- Kafka：高吞吐、分区内有序、消息可回放，适合日志流、事件溯源。
本地起三个实例最快的方式是 Docker：
docker run -d --name nats -p 4222:4222 -p 8222:8222 nats:latest -js
docker run -d --name rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3-management
docker run -d --name kafka -p 9092:9092 apache/kafka:3.7.0
注意：NATS 必须带 -js 参数才有 JetStream，否则后面的持久化 API 全部报错。
第二步：NATS 客户端（nats.go）
go get github.com/nats-io/nats.go
nc, err := nats.Connect("nats://localhost:4222",
nats.MaxReconnects(-1),
nats.ReconnectWait(2*time.Second),
)
if err != nil { log.Fatal(err) }
defer nc.Close()
// 发布
nc.Publish("orders.created", payload)
nc.Flush() // 确保写出去
// 订阅（支持通配符 * 和 >）
sub, _ := nc.Subscribe("orders.*", func(m *nats.Msg) {
log.Printf("收到: %s", m.Subject)
})
sub.SetPendingLimits(1024, 8<<20) // 防止慢消费者被静默丢弃
需要不丢消息就上 JetStream：
js, _ := nc.JetStream()
js.AddStream(&nats.StreamConfig{Name: "ORDERS", Subjects: []string{"orders.>"}})
js.Publish("orders.created", payload, nats.MsgId("order-1001")) // MsgId 做去重
sub, _ := js.PullSubscribe("orders.created", "worker-group", nats.BindStream("ORDERS"))
for {
msgs, _ := sub.Fetch(10, nats.MaxWait(5*time.Second))
for _, m := range msgs {
handle(m.Data)
m.Ack()
}
}
注意：核心 NATS 的 Publish 是 at-most-once，消费者不在线消息就没了。生产环境要可靠性，必须走 JetStream 或工作队列模式。
第三步：RabbitMQ 客户端（amqp091-go）
go get github.com/rabbitmq/amqp091-go
conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
ch, _ := conn.Channel()
ch.QueueDeclare("orders", true, false, false, false, nil)
ch.PublishWithContext(ctx, "", "orders", false, false, amqp.Publishing{
DeliveryMode: amqp.Persistent, // 持久化
ContentType: "application/json",
Body: payload,
})
消费端务必手动 ack，并限制预取数量：
ch.Qos(10, 0, false) // 每个消费者最多压 10 条未确认
msgs, _ := ch.Consume("orders", "", false, false, false, false, nil)
for d := range msgs {
if err := handle(d.Body); err != nil {
d.Nack(false, true) // 重新入队
continue
}
d.Ack(false)
}
注意：Consume 的第 3 个参数是 autoAck。设成 true 代表消息一投递就确认，进程崩溃消息直接丢失，业务消费一律用 false。
注意：amqp.Channel 不是并发安全的，多个 goroutine 同时发布要各用各的 channel，或者加锁。另外 amqp091-go 不做自动重连，得自己监听 conn.NotifyClose() 写重连循环。
第四步：Kafka 客户端（kafka-go / franz-go）
segmentio/kafka-go 上手最直观：
go get github.com/segmentio/kafka-go
w := &kafka.Writer{
Addr: kafka.TCP("localhost:9092"),
Topic: "orders",
Balancer: &kafka.LeastBytes{},
RequiredAcks: kafka.RequireAll, // 所有 ISR 确认
}
w.WriteMessages(ctx, kafka.Message{Key: []byte(orderID), Value: payload})
r := kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{"localhost:9092"},
GroupID: "order-workers",
Topic: "orders",
MinBytes: 1e3,
MaxBytes: 10e6,
CommitInterval: time.Second,
})
for {
m, err := r.FetchMessage(ctx)
if err != nil { break }
handle(m.Value)
r.CommitMessages(ctx, m)
}
如果追求性能，用 franz-go：kgo.NewClient(kgo.SeedBrokers(...))，默认就开启幂等生产，手动提交用 kgo.DisableAutoCommit() + cl.CommitRecords(ctx, r)。
注意：Kafka 只保证分区内有序。要保证同一订单的消息顺序，必须把订单号作为 Key，让它落到同一分区。
第五步：统一封装与优雅退出
三种 MQ 抽一层接口，业务代码就能换实现：
type Publisher interface {
Publish(ctx context.Context, topic, key string, body []byte) error
}
type Subscriber interface {
Subscribe(ctx context.Context, topic string, h func(context.Context, []byte) error) error
}
退出时用 signal.NotifyContext 收 SIGTERM，然后：停订阅 → 等 in-flight 消息处理完（sync.WaitGroup）→ 关连接。别在 handler 里直接 os.Exit，否则正在处理的消息会被重复投递。
第六步：通用踩坑清单
- 幂等：三家的可靠投递都是「至少一次」，重复消费是常态，业务侧要有唯一键去重（订单号、消息 ID）。
- 慢消费者：NATS 有 pending 上限、RabbitMQ 有 Qos、Kafka 有
MaxBytes，都要设，否则内存先爆。
- 超时：所有
Publish / Fetch 都传 context.WithTimeout，别用 context.Background() 裸跑。
- 健康检查：Exporter 暴露连接状态，重连风暴时能第一时间看到。
小结
- NATS 快而轻，要可靠性开 JetStream，用
MsgId 去重、PullSubscribe + Fetch + Ack。
- RabbitMQ 路由灵活，
QueueDeclare 持久化 + DeliveryMode: Persistent + 手动 ack + Qos 预取是标配；Channel 非并发安全，重连要自己写。
- Kafka 用 Key 控分区顺序，
RequireAll 保证写入可靠，消费端手动提交 offset。
- 三者共同点：接口抽象、context 超时、优雅关闭、业务幂等——这四条做好，换 MQ 的成本就很低。
正式会员
认证极客
钻石卡会员
见习用户
