RocketMQ 消息堆积告警：消费线程数与批量拉取的 4 个参数

我不会起名字322· 后端 / 算法 / 数据库 技术栈 力扣 Hot100 Go 项目 Redis MySQL文章目录RocketMQ 消息堆积告警消费线程数与批量拉取的 4 个参数一、先分清真堆积和假堆积二、参数 1消费线程数三、参数 2批量拉取条数四、参数 3流控阈值五、参数 4消费超时与重试六、带走物RocketMQ 消息堆积告警消费线程数与批量拉取的 4 个参数一、先分清真堆积和假堆积RocketMQ 控制台里的堆积量是堆积量 最大位点(maxOffset) - 消费位点(consumerOffset)这个值上涨有两种完全不同的原因真堆积生产速度 消费速度差值持续扩大。假堆积消费在正常跑但某几条消息一直消费失败卡住了位点导致差值看起来很大——实际上后面几万条早就消费完了。区分方法# 看消费位点是不是在动shmqadmin consumerProgress-nlocalhost:9876-gorder-consumer-group# 隔 10 秒看两次对比 consumerOffset位点在涨 → 真堆积去查消费能力位点纹丝不动→ 卡在某条消息上去查失败重试。后者非常常见一条消息因为反序列化失败/业务异常反复重试RocketMQ 会把它丢进重试队列但如果重试次数用尽进了死信队列而你没消费死信队列位点就会一直卡着。先查死信队列有没有积压# 死信 topic 的命名规则%DLQ%consumer-group二、参数 1消费线程数RocketMQ 的消费者有两个线程数参数很多人只调了其中一个// PushConsumerconsumer.setConsumeThreadMin(20);// 最小消费线程consumer.setConsumeThreadMax(64);// 最大消费线程它们的含义ConsumeThreadMin默认 20常驻线程数ConsumeThreadMax默认 64峰值时最多扩到多少。但线程数不是越大越好它受一个硬约束一个消费队列MessageQueue在同一时刻只能被一个线程消费。topic 有 16 个队列你能消费的队列数受限于 - 消费者实例数 × 每个实例分到的队列数 - 单实例下线程数 分到的队列数时多余的线程是空闲的所以正确的扩容顺序是1. 先看队列数够不够ReadQueueNums 2. 队列数不够 → 先扩队列但队列只能加不能减要慎重 3. 队列数够了 → 再调线程数# 看 topic 的队列数shmqadmin topicRoute-nlocalhost:9876-torder-topic只调线程数不调队列数是 RocketMQ 里最常见的无效优化。另外注意如果消费逻辑里有同步 IO调下游接口、写库线程数应该设得比 CPU 核数大得多比如 64~200因为线程大部分时间在等 IO如果是纯 CPU 计算线程数超过核数反而会增加上下文切换开销。三、参数 2批量拉取条数// 每次从 broker 拉多少条默认 32PushConsumer 内部使用consumer.setPullBatchSize(32);// 每次交给业务线程处理的条数默认 1consumer.setConsumeMessageBatchMaxSize(1);这两个参数经常被混为一谈参数管的是影响PullBatchSize一次网络拉取的条数网络往返次数ConsumeMessageBatchMaxSize一次交给MessageListener的条数业务批处理的机会堆积场景下真正该调的是ConsumeMessageBatchMaxSize。从 1 调到 10意味着业务层可以做批量落库consumer.registerMessageListener((MessageListenerConcurrently)(msgs,context)-{// msgs.size() 最大就是 ConsumeMessageBatchMaxSizeListOrderordersmsgs.stream().map(this::parse).collect(toList());orderMapper.batchInsert(orders);// 一次 insert 100 条returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;});批量落库和逐条落库的差距通常是一个数量级100 条一次 insert vs 100 次 insert。但要小心批量变大后如果批次里有一条失败默认会整批重试。所以要么业务层自己做容错成功的跳过要么把批量控制在可接受的范围。另外PullBatchSize调大时要注意单条消息大小PullBatchSize × 单条消息大小 不能超过 broker 的 maxTransferBytesOnMessageInMemory默认 256KB消息体大的话调大PullBatchSize会报拉取失败。四、参数 3流控阈值RocketMQ 客户端有自动流控机制堆积太多时它会主动降速——这经常被人误判成消费能力不行。触发条件默认1. 本地缓存消息数 pullThresholdForQueue默认 1000 2. 本地缓存消息大小 pullThresholdSizeForQueue默认 100MB 3. 本地缓存消息跨度 consumeConcurrentlyMaxSpan默认 2000触发后 broker 会返回流控信息客户端暂停拉取等缓存消化。// 堆积场景可以适当放宽consumer.setPullThresholdForQueue(2000);consumer.setPullThresholdSizeForQueue(200);// 单位 MB判断是不是被流控了看客户端日志里有没有so do flow control, and the flow control threshold is 1000有的话说明不是消费慢而是本地缓存满了在限流。这时调大阈值往往比调线程数更有效。五、参数 4消费超时与重试// 单条消息消费的最大时长默认 15 分钟consumer.setConsumeTimeout(15);超时后 RocketMQ 会把它当失败处理进重试队列。注意这个时间很长15 分钟如果你的业务卡住了位点会一直不动看起来像完全没在消费。重试策略并发消费第 1 次失败 → 延迟 10s 重试level 3 第 2 次失败 → 延迟 30slevel 4 ...逐级递增最多 16 次 16 次都失败 → 进死信队列 %DLQ%所以堆积排查一定要先看重试队列和死信队列那里往往藏着真正的元凶——一条消息反复重试会持续占用消费资源。六、带走物堆积告警的处理顺序1. 看消费位点动不动 ├─ 不动 → 卡在某条失败消息上查重试队列 死信队列 %DLQ% └─ 在动但上涨 → 真堆积继续第 2 步 2. 看 topic 队列数ReadQueueNums └─ 队列数 消费者实例数 → 先扩队列线程数调再大也没用 3. 调 ConsumeMessageBatchMaxSize默认 1 太小 └─ 让业务层能批量落库这是收益最大的一项 4. 查客户端日志有没有 do flow control └─ 有 → 调大 pullThresholdForQueue / pullThresholdSizeForQueue 5. 最后才调 ConsumeThreadMax └─ 同步 IO 型可以设大64~200纯 CPU 型别超过核数太多一句话结论RocketMQ 堆积时先查位点卡没卡和队列够不够这两项没查就调线程数是最浪费时间的做法。