当前位置：首页 > news >正文

Go 队列消费者：并发数不是配置越大越能扛

news 2026/7/6 12:09:45

Go 队列消费者：并发数不是配置越大越能扛

一、消费者扩太猛，会把下游一起拖下水

队列系统常被当成削峰工具。请求进队列，消费者慢慢处理，看起来很稳。但消费者并发数如果随便拉大，就会把数据库、第三方 API、对象存储一起打满。队列只是把压力换了个地方，不会凭空消灭压力。

消费者设计要按下游容量反推并发。吞吐、失败率、重试策略、单任务耗时都要一起看。并发越大，未必越快，可能只是失败得更快。

我处理过一起典型事故：监控看到消息队列积压 80 万条，运维第一反应是"加消费者"。消费者并发从 50 拉到 200，CPU 和内存都不是问题。但五分钟之后数据库连接池打满——每条消息处理都要查两次数据库、写一次，200 个并发意味着可能同时有 600 个数据库请求。连接池只有 100 个连接，瞬间耗尽。然后消息开始大量 ack 超时，回到队列，消费者重新拉取，造成重复消费，数据库写入冲突更多。加消费者的结果是：积压没减少，数据库差点被打挂。真正的解决方案是先限流队列的推送速率，等数据库连接池恢复后，再逐步放开消费者。

二、消费者链路要包含限流、重试和死信

一个生产消费者至少要有并发限制、处理超时、重试次数和死信队列。

flowchart TD A[消息队列] --> B[消费者拉取] B --> C[并发限制] C --> D[处理任务] D --> E{成功} E -->|是| F[确认消息] E -->|否| G{可重试} G -->|是| H[延迟重试] G -->|否| I[死信队列]

没有死信队列，坏消息会在队列里无限转圈。系统看起来很努力，实际在原地打滑。

重试策略不能是简单的"失败就等一下再试"。重试次数要有限制（通常 3-5 次），重试间隔要指数退避（1s → 2s → 4s → 8s），并且区分可重试错误和不可重试错误。参数校验失败不应该重试，下游临时超时可以重试。重试策略写清后，队列的"假忙碌"会大幅减少。

三、用 semaphore 控制本地并发

下面示例用信号量限制处理并发。真实项目还要处理 ack、nack 和 shutdown。

func Consume(ctx context.Context, msgs <-chan Message, limit int, handle func(context.Context, Message) error) { sem := make(chan struct{}, limit) for { select { case <-ctx.Done(): return case msg := <-msgs: sem <- struct{}{} go func(m Message) { defer func() { <-sem }() _ = handle(ctx, m) }(msg) } } }

这里的limit不是随便拍。要根据下游能力、任务耗时和机器资源压测出来。压测时不仅要看"多少并发时下游 CPU 到 80%"，还要看"并发每增加 20%，P99 延迟增加多少"。延迟增长到不可接受时就应该是并发上限，而不是等 CPU 打满才停。

另外，处理超时要独立于并发限制。并发数只控制同时处理的数量，但如果某些消息处理超过预期时间（比如下游慢查询），就应该主动超时取消。goroutine 泄漏最常见的来源就是消费者处理超时后没有及时退出。