直接用nsqConsumer.HandleMessage会炸服务，因客户端不控并发，每消息启goroutine致突发流量下内存飙升、调度卡死、GC频繁而OOM；MaxInFlight仅控未确认消息数，非实际并发数，须配合channel+worker pool限流。
为什么直接用 go nsqConsumer.HandleMessage 会炸掉服务
NSQ 客户端默认不控制消费并发，HandleMessage 回调一来就起 goroutine 处理，遇上突发流量（比如 1000 条/s 进来），瞬间 spawn 几千个 goroutine —— 内存飙升、调度器卡死、GC 频繁，最后 OOM 或响应延迟飙升。这不是 NSQ 的问题，是没做并发节流。
- NSQ 客户端本身不提供内置的 worker pool，
nsq.Consumer的MaxInFlight只控制「已发未确认」消息数，不等于并发处理数 -
MaxInFlight设太高（如 5000）+ 消息处理慢（如含 HTTP 调用），会导致大量消息堆积在内存里，触发 GC 压力 - 设太低（如 1）又压不住吞吐，实际并发还是靠你手动控制 goroutine 启动节奏
nsq.Consumer 的 MaxInFlight 和真实并发的关系
MaxInFlight 是 NSQ 协议层的流控参数：它限制「已推送给客户端但尚未 finish 或 requeue」的消息总数。它影响的是 NSQ 服务器是否继续推送新消息，不是你本地启动几个 goroutine 的开关。
- 假设
MaxInFlight = 100，且每条消息平均处理耗时 200ms，则理论最大并发 ≈ 100 / 0.2 = 500（前提是你的 handler 真的能立刻启 goroutine 并跑满） - 但如果你 handler 里没加限流，100 条消息进来瞬间就启 100 个 goroutine；若其中 20 条卡住（比如下游超时），剩下 80 条还在等，
MaxInFlight就被占满，NSQ 暂停派发，吞吐断崖下跌 - 正确姿势是：把
MaxInFlight设为略大于你期望的 worker 数（比如 worker=20 →MaxInFlight=25），再用 channel + worker pool 控制实际并发
用 channel + 固定 worker pool 接管消息处理
别在 HandleMessage 回调里直接干活，把它变成「投递任务」的动作，让一组固定数量的 worker 从共享 channel 拉取并执行。
// 初始化 worker pool
msgChan := make(chan *nsq.Message, 1000)
for i := 0; i < 20; i++ {
go func() {
for msg := range msgChan {
// 实际业务逻辑放这里
if err := processOrder(msg.Body); err != nil {
msg.Requeue(-1) // 失败重入队
continue
}
msg.Finish()
}
}()
}
<p>// 注册 NSQ handler：只负责转发，不阻塞
consumer.AddHandler(nsq.HandlerFunc(func(msg *nsq.Message) error {
select {
case msgChan <- msg:
return nil
default:
// channel 满了，主动拒绝（避免堆积）
return nsq.Errorf("msg chan full")
}
}))
-
msgChan缓冲区大小要合理：太小（如 10）易丢拒收；太大（如 10000）可能掩盖下游瓶颈 - worker 数量建议从 5–20 起步，根据单条消息平均耗时和 CPU 核数调整（例如 4 核机器，平均处理 150ms，20 个 worker 已接近饱和）
- 务必在
processOrder中捕获 panic，否则一个 panic 会让整个 worker goroutine 退出，可用recover()包一层
Gin 启动时如何安全绑定 NSQ consumer
Gin 的 router.Run() 是阻塞调用，NSQ consumer 必须提前启动，并确保 consumer 实例在 Gin server 崩溃重启时也能优雅关闭。
- 不要在
main()里先router.Run()再启 consumer —— Gin 卡住后 consumer 根本不会运行 - 用
sync.WaitGroup或errgroup.Group统一管理多个长期运行的 goroutine（Gin server + NSQ consumer） - 监听 OS 信号（
os.Interrupt,syscall.SIGTERM），收到后先调用consumer.Stop()，等待consumer.Stop()返回（它会等所有消息处理完或超时）再退出 - 注意：NSQ consumer 的
Stop()默认有 30s 超时，可通过consumer.SetLogger(...)查看是否因消息迟迟不Finish()导致强制退出
NSQ 本身不维护状态，consumer 崩溃重启后会从最新 offset 拉消息，所以关键是要保证每条消息的 Finish() 或 Requeue() 调用不遗漏 —— 这比“高并发”更难，也更容易被忽略。
