★ FEATURED ARTICLE 1. 这不是“学完就能用”的教程而是我在三个高并发系统里踩出来的RocketMQ使用地图你点开这个标题大概率不是想看“消息队列是什么”这种教科书定义——你手头正压着一个订单超时未支付要自动关单、用户行为日志要异步落库、或者跨服务库存扣减必须强一致的活儿。你查到RocketMQ是因为它在阿里系、金融、电商场景里被反复验证过扛得住每秒十万级消息堆积支持事务消息保最终一致延迟消息精度能控在毫秒级。但真正上手后你会发现官方文档里那个DefaultMQProducer.send()调用背后藏着一堆不写进API但决定你系统生死的细节比如send()方法默认是同步发还是异步发retryTimesWhenSendFailed设成3次到底重试的是网络超时、Broker宕机还是主从切换间隙为什么测试环境跑得好好的一上生产就出现消息重复消费而你的幂等逻辑偏偏漏掉了数据库唯一索引这个最基础的兜底我过去三年主导过三个不同量级的RocketMQ落地项目一个日均订单200万的电商平台做订单状态流转解耦一个实时风控系统做规则引擎与决策中心的事件分发还有一个IoT平台处理千万级设备心跳上报。这三个项目没一个是在“标准配置”下跑通的——它们各自卡在不同的点上第一个项目上线第三天因Broker集群磁盘IO打满导致消息积压消费者拉取超时第二个项目在灰度阶段发现事务消息回查接口被高频触发拖垮了下游账户服务第三个则在压测时暴露出批量消费线程池配置不当导致设备离线告警延迟高达47秒。这些坑不会出现在任何一份QuickStart文档里但会真实地让你凌晨三点被电话叫醒。这篇内容就是把这三套系统里拆解出的核心编程模型认知框架和生产环境不可妥协的硬性约束掰开揉碎讲清楚。它不教你如何安装NameServer但会告诉你MessageQueueSelector的实现里为什么绝对不能用System.currentTimeMillis()做哈希因子它不罗列所有配置项但会算给你看当你的Topic有16个队列、消费者组有8个实例时每个实例实际分配到几个队列以及这个数字怎么直接影响你的消费吞吐瓶颈。如果你正在为消息丢失、重复、延迟、堆积发愁或者刚接手一个别人写的RocketMQ模块却看不懂为什么这么设计——那接下来的内容就是你该逐行抄进自己笔记里的实操地图。2. 核心编程模型不是API列表而是四层责任边界的动态平衡很多人把RocketMQ编程模型简单理解为“生产者发消息、消费者收消息”这就像把开车说成“踩油门、踩刹车”。真正的模型复杂性藏在消息生命周期内四个关键角色的责任划分与协作契约中。这四层不是静态结构而是随着流量、故障、配置变化持续动态博弈的系统。我把它拆成 Producer端、Broker端、Consumer端、运维治理端 四个视角每个视角都对应一套必须理解的底层契约。2.1 Producer端发送不是“发出去就完事”而是三次握手式的责任移交Producer的send()方法表面看只返回一个SendResult但这个返回值背后是Producer与Broker之间完成了一次完整的“责任移交确认”。这个确认过程分三步走每一步失败的处理逻辑完全不同第一步网络可达性确认Producer先尝试与Broker建立TCP连接。如果Broker地址配置错误或网络策略阻断会直接抛MQClientException错误码10001无法连接到Broker。这里的关键陷阱是很多团队用DNS域名配置Broker地址却忽略了K8s Service的DNS缓存TTL。当Broker Pod滚动更新时旧IP可能被客户端缓存长达30秒导致这段时间内所有发送请求都失败。解决方案不是加重试而是强制设置JVM参数-Dsun.net.inetaddr.ttl1让DNS解析结果只缓存1秒。第二步Broker接收确认连接建立后Producer将消息序列化为二进制包通过Netty Channel发送给Broker。Broker收到后会先校验消息体大小默认上限4MB、Topic是否存在、权限是否合法。如果校验失败Broker立即返回RemotingCommand响应包Producer解析后抛MQBrokerException错误码如10002Topic不存在或10003权限不足。注意此时消息根本没进Broker存储层所以不会产生任何磁盘IO。很多团队误以为“发送失败消息丢了”其实只是Broker拒绝了这次移交请求。第三步存储落盘确认可选但关键这才是决定消息是否“真正可靠”的一步。RocketMQ提供三种发送模式SEND_ASYNCProducer发完包就返回不等Broker落盘响应。吞吐最高但Broker崩溃时可能丢失。SEND_SYNCProducer阻塞等待Broker返回SEND_OK且Broker必须确保消息已写入CommitLog即刷盘成功。这是默认模式也是生产环境唯一推荐的模式。SEND_ONEWAY完全不等响应连网络异常都不感知。仅用于日志类非关键消息。提示SEND_SYNC模式下Broker的刷盘策略flushDiskTypeASYNC_FLUSH或SYNC_FLUSH决定了最终可靠性。ASYNC_FLUSH是默认值Broker将消息写入PageCache后即返回成功依赖Linux内核定时刷盘通常1秒一次。这意味着极端情况下如服务器突然断电最多丢失1秒内的消息。如果你的业务要求“零丢失”必须改为SYNC_FLUSH但代价是磁盘IO压力翻倍TPS下降约30%。我们曾在一个支付对账系统中强制启用SYNC_FLUSH结果发现单台Broker磁盘IOPS峰值突破12000不得不将Topic拆分到更多Broker节点分摊压力。2.2 Broker端不是被动存储而是消息生命周期的仲裁者Broker常被误解为“消息硬盘”但它真正的核心职责是仲裁消息在不同状态间的转换。一条消息在Broker内部经历五个关键状态PREPARED预提交→COMMIT已提交→CONSUMED已消费→DELETED已删除→EXPIRED已过期。其中前两个状态专属于事务消息后三个是普通消息的通用状态。事务消息的仲裁逻辑当Producer发送事务消息时Broker先将其标记为PREPARED状态并存入CommitLog但不向Consumer暴露。此时Producer执行本地事务如扣减库存然后根据事务结果调用commitMessage()或rollbackMessage()。Broker收到指令后才将消息状态更新为COMMIT或DELETED。但如果Producer在执行本地事务后崩溃Broker无法收到指令就会触发事务回查机制Broker每隔60秒默认向Producer发起checkLocalTransactionState()回调询问该消息关联的本地事务状态。这个回查不是无限次的——默认最多查15次之后自动丢弃PREPARED消息。问题来了如果Producer的回查接口因为数据库连接池耗尽而超时Broker会认为事务失败导致本该成功的消息被回滚。我们在风控系统中就遇到过这个问题回查接口依赖MySQL而MySQL连接池最大连接数设为20但回查线程池默认是20个线程高峰期所有线程都在等DB连接形成死锁。解决方案是将回查线程池独立配置transactionCheckInterval30000transactionCheckMax5并确保其连接池与业务连接池物理隔离。消息过期的物理实现RocketMQ没有为每条消息单独维护过期时间戳而是采用时间轮文件级清理的组合策略。所有消息按写入时间顺序追加到CommitLog文件中每个CommitLog文件固定1GB。Broker启动时会扫描所有CommitLog文件计算每个文件的最后修改时间。如果某个文件的最后修改时间早于fileReservedTime48小时且该文件中所有消息的storeTimestamp都已超过messageDelayLevel配置的延迟时间则整个文件被标记为可删除。这意味着即使你给某条消息设置了2小时延迟只要它所在的CommitLog文件还没满就不会被清理而一旦文件满了被滚动里面所有消息无论是否到期都会被整体移出有效范围。这就是为什么在低流量场景下延迟消息的实际触发时间可能比设定值多出几分钟——它在等文件滚动。2.3 Consumer端消费不是“拿过来就处理”而是状态机驱动的协作协议Consumer的MessageListenerConcurrently接口看似简单但它的执行模型是一个严格的状态机。每次拉取到消息后Consumer会按以下流程处理将消息放入本地内存队列ProcessQueue启动线程池consumeThreadMin20从队列中取消息调用listener.consumeMessage()处理处理成功后向Broker提交消费位点offset如果处理失败按consumeMessageBatchMaxSize1默认逐条重试这个流程里最致命的陷阱在第4步消费位点提交不是原子操作。Consumer先在本地内存更新offset再异步发送UpdateConsumerOffsetRequest给Broker。如果Consumer进程在发送请求前崩溃Broker记录的offset就比实际消费位置落后导致重启后重复消费。我们电商系统的订单关单服务就因此重复发送了37次关单指令幸好下游有基于订单号的幂等校验。但更隐蔽的问题是当Consumer配置了pullBatchSize32一次拉32条而consumeThreadMax10最多10个线程并发处理时如果某条消息处理耗时2秒其他31条消息会被卡在内存队列里等待造成人为的消费延迟。解决方案是将pullBatchSize设为1用consumeMessageBatchMaxSize1保证单条处理虽然吞吐略降但延迟可控。2.4 运维治理端监控不是“看图表”而是对四个黄金指标的实时干预生产环境的RocketMQ治理本质是对四个不可妥协的黄金指标的持续盯防消息堆积量InSyncReplicas不是看brokerOffset - consumerOffset的差值而是看InSyncReplicas数量。当这个值小于minInsyncReplicas2默认时说明Follower副本同步滞后Broker会拒绝写入新消息防止脑裂。我们曾因网络抖动导致InSyncReplicas从3降到1整个Topic写入阻塞订单创建接口超时率飙升至90%。消费延迟CONSUME_LAGBroker统计的是lastStoreTime - lastConsumeTime但这个值有误导性。真正的业务延迟要看consumerOffset对应的storeTimestamp与当前时间的差值。我们用Prometheus采集rocketmq_consumer_lag_seconds指标当连续5分钟60秒时自动触发告警并扩容Consumer实例。Broker负载dispatchMaxBuffer这是CommitLog文件写入PageCache的缓冲区大小默认1G。当缓冲区满时Broker会强制刷盘导致写入延迟毛刺。我们通过jstat -gc监控S0C/S1CSurvivor区容量和ECEden区容量当EC使用率持续90%说明GC频繁间接反映缓冲区压力大需调整dispatchMaxBuffer。网络连接数clientChannelCount每个Producer/Consumer连接占用一个Netty Channel。当连接数超过nettyServerSocketChannelMax5000默认新连接会被拒绝。我们曾因Consumer实例未正确关闭连接导致连接数缓慢增长两周后达到4999新Consumer无法接入。解决方案是启用channelCloseIdletrue自动关闭空闲连接。3. 生产环境最佳实践不是配置清单而是七条血泪换来的硬性红线所谓“最佳实践”不是教你把所有参数调到最大而是明确哪些配置绝对不能碰、哪些操作必须做、哪些场景必须规避。以下是我在三个项目中用故障换来的七条不可逾越的红线。3.1 红线一Topic命名必须带业务域前缀且禁止跨域复用Topic我们第一个项目曾为图省事用一个topic_order承载所有订单相关消息创建、支付、发货、退货。结果在一次大促压测中退货消息量激增占满所有队列带宽导致新订单创建消息延迟超10分钟。根本原因是RocketMQ的队列MessageQueue是Topic级别的资源所有消息共享同一组队列。当某个子业务突发流量会挤占其他子业务的消费能力。解决方案是按业务域拆分Topictopic_order_create、topic_order_pay、topic_order_ship。每个Topic独立配置队列数defaultTopicQueueNums16这样退货服务扩容时只需增加topic_order_return的Consumer实例不影响其他Topic。注意Topic名不能包含下划线_或点号.因为RocketMQ内部用这些字符做路由分隔。我们曾用topic.order.pay结果Broker解析时误判为topic和order.pay两个层级导致路由失败。3.2 红线二Consumer Group必须与业务场景强绑定禁止多个业务共用GroupConsumer Group的本质是消费位点offset的归属单位。同一个Group下的所有Consumer实例共享一个消费位点。如果订单服务和物流服务共用group_order那么当物流服务消费到第1000条消息时订单服务重启后会从第1000条开始消费跳过前面所有订单创建消息我们第二个项目就因此导致大量订单状态不同步。正确做法是每个业务场景独占Group命名规则为{业务域}_{环境}_{功能}如order_prod_create、logistics_uat_ship。3.3 红线三消息体必须小于4MB且禁止在消息中传递大对象引用RocketMQ默认单条消息最大4MB这是硬编码在MAX_MESSAGE_SIZE4194304中的。但更危险的是“逻辑超限”比如你在消息体里放一个JSON里面包含用户完整画像含图片base64虽然JSON字符串长度没超4MB但反序列化后Java对象内存占用可能达20MB直接OOM。我们的IoT平台曾因设备上报的固件版本信息里嵌入了二进制指纹导致Consumer JVM堆内存瞬间打满。解决方案是所有大于100KB的数据必须存OSS/S3消息体里只传URL和MD5校验值。3.4 红线四事务消息的本地事务必须是幂等的且回查接口必须无副作用事务消息的checkLocalTransactionState()接口必须满足两个条件第一查询逻辑不能修改任何数据无副作用第二查询结果必须稳定幂等。我们风控系统最初在回查接口里做了“如果未查到记录就插入一条默认风控结果”结果Broker多次回查导致同一条消息生成了多条风控记录。正确做法是回查只查数据库transaction_state字段该字段由主事务执行时写入回查绝不写入。3.5 红线五延迟消息等级必须严格匹配业务需求禁止滥用DELAY_LEVELRocketMQ的延迟消息通过delayLevel参数控制取值1-18对应时间如下1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h注意没有“15秒”、“45分钟”这种精确值。如果你的业务需要“15秒后触发”必须选delayLevel430秒然后在Consumer端用System.nanoTime()计算实际延迟不足15秒就休眠补足。我们曾因强行用delayLevel310秒去凑15秒结果大量消息在10秒时被消费业务逻辑失败。3.6 红线六Consumer必须实现shutdown()优雅退出且禁止在监听器中阻塞Consumer进程退出时必须调用consumer.shutdown()否则offset不会提交到Broker重启后会重复消费。更严重的是如果在MessageListenerConcurrently.consumeMessage()里写了Thread.sleep(10000)会导致整个Consumer线程池被占满新消息无法处理。我们电商系统曾因日志打印里加了FileWriter.append()同步IO导致单条消息处理耗时从20ms飙升到8秒线程池全堵死。解决方案是所有IO操作必须异步化用CompletableFuture.runAsync()包装。3.7 红线七Broker磁盘必须用SSD且storePathCommitLog与storePathConsumeQueue必须分盘CommitLog是消息的原始存储写入是顺序IOConsumeQueue是消息索引读取是随机IO。如果两者放在同一块机械硬盘上顺序写和随机读会互相干扰磁盘IO util持续100%消息堆积雪崩。我们第一个项目就因此在大促期间磁盘IO打满。必须将storePathCommitLog挂载到SSD盘如/data1/rocketmq/store/commitlogstorePathConsumeQueue挂载到另一块SSD如/data2/rocketmq/store/consumequeue并通过df -h持续监控两块盘的使用率当任一盘85%时自动告警。4. 实操过程从零搭建一个抗住百万QPS的订单消息链路现在我们把前面所有原则落地到一个具体场景电商订单创建后的状态通知链路。目标是支撑大促期间峰值120万QPS的订单创建请求消息端到端延迟200ms堆积容忍度1亿条。4.1 架构设计三层解耦每层可独立伸缩整个链路分为三层接入层Nginx Spring Cloud Gateway负责HTTP请求接入、限流Sentinel QPS120w、熔断消息层RocketMQ集群3个NameServer6个Broker3主3从每个Broker配16核32G2TB SSD消费层订单状态服务Java Spring Boot部署在K8sHPA根据rocketmq_consumer_lag_seconds指标自动扩缩容关键设计点Topic命名为topic_order_create队列数设为64defaultTopicQueueNums64因为6个Broker * 每Broker 16个队列 96但实际可用队列数取最小公倍数64能被8、16、32整除方便Consumer实例数扩展。Producer使用DefaultMQProducersendMsgTimeout30003秒超时retryTimesWhenSendFailed2失败重试2次避免网络抖动误判。Consumer Group命名为group_order_create_prodconsumeMessageBatchMaxSize1单条处理避免延迟放大pullBatchSize1一次只拉1条精准控速。4.2 Broker集群部署参数调优的物理依据6个Broker节点配置broker.conf关键参数# 磁盘空间保障 diskMaxUsedSpaceRatio85 # 磁盘使用率超85%时拒绝写入 fileReservedTime72 # CommitLog文件保留72小时覆盖大促后3天数据 # 性能调优 flushDiskTypeASYNC_FLUSH # 不用SYNC_FLUSH用SSD足够冗余保可靠性 flushIntervalCommitLog1000 # 每秒刷盘一次平衡性能与安全 mapedFileSizeCommitLog1073741824 # 1GB CommitLog文件避免小文件过多 # 内存优化 transientStorePoolEnabletrue # 启用堆外内存池减少GC压力 transientStorePoolSize5 # 堆外内存池5G应对突发流量为什么这样配因为SSD的随机读写IOPS可达30000而机械盘仅150。ASYNC_FLUSH配合SSD刷盘延迟稳定在0.2ms内远低于SYNC_FLUSH的1.5ms。transientStorePoolEnabletrue将CommitLog写入从堆内存移到堆外内存避免大促时Full GC导致消息写入停顿。我们实测过不开堆外内存池时GC pause平均120ms开启后GC pause降至3ms以内。4.3 Producer代码不只是发消息更是流量整形器订单服务的Producer初始化代码public class OrderProducer { private DefaultMQProducer producer; public void init() throws MQClientException { producer new DefaultMQProducer(producer_order_create); producer.setNamesrvAddr(ns1:9876;ns2:9876;ns3:9876); producer.setSendMsgTimeout(3000); producer.setRetryTimesWhenSendFailed(2); // 关键启用VIP通道绕过NameServer路由直连Broker producer.setVipChannelEnabled(false); producer.start(); } public SendResult sendOrderCreate(Order order) throws MQException { try { Message msg new Message( topic_order_create, tag_order_create, order.toJson().getBytes(RemotingHelper.DEFAULT_CHARSET) ); // 关键设置Key为订单号便于问题追踪 msg.setKeys(order.getOrderId()); // 关键设置延迟等级为11秒后投递给下游留出DB写入时间 msg.setDelayTimeLevel(1); return producer.send(msg, (mqs, msg1, arg) - { // 消息队列选择器按订单号哈希保证同一订单消息路由到同一队列 int queueIndex Math.abs(order.getOrderId().hashCode()) % mqs.size(); return mqs.get(queueIndex); }); } catch (Exception e) { // 关键捕获所有异常记录详细日志绝不让异常穿透到业务层 log.error(send order create failed, orderId{}, order.getOrderId(), e); throw new MQException(send order create failed, e); } } }这里三个关键点setVipChannelEnabled(false)禁用VIP通道。因为VIP通道会强制走NameServer做二次路由增加20ms延迟。直连Broker更稳更快。msg.setDelayTimeLevel(1)设1秒延迟不是为了“等”而是给订单主表DB写入留出时间窗口避免Consumer查不到刚创建的订单。队列选择器用orderId.hashCode()保证同一订单的所有消息如创建、支付落在同一队列Consumer按顺序消费状态流转不乱序。4.4 Consumer代码不是处理消息而是构建状态机订单状态服务的Consumerpublic class OrderStatusConsumer { private DefaultMQPushConsumer consumer; public void init() throws MQClientException { consumer new DefaultMQPushConsumer(group_order_create_prod); consumer.setNamesrvAddr(ns1:9876;ns2:9876;ns3:9876); consumer.setConsumeThreadMin(20); consumer.setConsumeThreadMax(50); // 大促时自动扩容到50线程 consumer.setPullBatchSize(1); // 一次只拉1条 consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) - { ListConsumeOrderlyStatus results new ArrayList(); for (MessageExt msg : msgs) { try { Order order JSON.parseObject(new String(msg.getBody()), Order.class); // 关键幂等校验先查DB存在则跳过 if (orderService.existsByOrderId(order.getOrderId())) { results.add(ConsumeOrderlyStatus.SUCCESS); continue; } // 关键DB操作必须在try-catch内失败则重试 orderService.createOrderStatus(order); results.add(ConsumeOrderlyStatus.SUCCESS); } catch (Exception e) { log.error(consume order create failed, orderId{}, new String(msg.getBody()), e); // 关键失败时返回SUSPEND_CURRENT_QUEUE_A_MOMENT暂停当前队列10秒 results.add(ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT); } } return results; }); consumer.subscribe(topic_order_create, *); consumer.start(); } }核心逻辑ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT当某条消息处理失败时不是立即重试可能DB还在恢复而是暂停整个队列10秒给下游系统恢复时间。我们实测过这个策略比盲目重试降低90%的重复消费。幂等校验orderService.existsByOrderId()必须走数据库主键查询不能用Redis缓存因为缓存可能有延迟导致误判。setConsumeThreadMax(50)K8s HPA根据rocketmq_consumer_lag_seconds指标当延迟10秒时自动将Pod副本数从10扩到50线程池随之扩容。4.5 压测验证不是跑通就行而是验证四维指标我们用JMeter模拟120万QPS订单创建请求持续30分钟监控四维指标Producer发送成功率99.998%失败0.002%全是网络瞬断重试后成功Broker消息堆积量峰值1200万条1亿容忍阈值30分钟后回落至0Consumer端到端延迟P99186ms200ms目标Broker磁盘IO util稳定在65%SSD承受范围内压测中发现一个关键问题当Consumer实例从10扩到50时InSyncReplicas从3降到2因为新扩容的Consumer实例拉取速度跟不上导致Follower副本同步滞后。解决方案是在K8s HPA策略中加入scaleUpCooldownSeconds300扩容冷却5分钟避免短时流量尖峰引发频繁扩缩容。5. 常见问题与排查技巧实录那些凌晨三点的电话教会我的事所有最佳实践都源于故障。我把过去三年处理过的典型问题整理成一张速查表并附上独家排查技巧。这些问题90%的RocketMQ文档都不会写但你一定会遇到。问题现象根本原因排查命令/工具解决方案我的实操心得消息堆积持续增长Consumer日志无报错Consumer线程池耗尽新消息无法进入处理队列jstack pid查看ConsumeMessageThread_线程数若全部WAITING说明队列满增加consumeThreadMax或检查pullBatchSize是否过大导致单条处理慢别急着扩容先jstack看线程状态。我们曾因此发现Consumer里有个while(true)死循环没加breakConsumer重启后重复消费大量消息offset未提交成功Broker记录的位点落后于实际消费位置mqadmin consumerProgress -g group_name -n nameserver_addr查看brokerOffset和consumerOffset差值确保Consumer调用shutdown()并在finally块中强制提交offset在Consumer的PreDestroy方法里加consumer.shutdown()比Runtime.getRuntime().addShutdownHook()更可靠Broker磁盘IO util 100%消息写入延迟毛刺严重dispatchMaxBuffer缓冲区满强制刷盘cat /proc/broker_pid/status | grep VmRSS查看Broker内存占用若30G说明缓冲区压力大增加transientStorePoolSize或升级SSD监控VmRSS比监控heap更有价值因为RocketMQ大量使用堆外内存事务消息回查接口被高频触发下游服务被打垮Producer本地事务执行慢超时后Broker发起回查mqadmin topicStatus -t topic_name -n nameserver_addr查看transactionCheckTimes字段优化本地事务SQL加索引将回查线程池独立配置transactionCheckMax3回查不是救火是预警。当transactionCheckTimes100说明本地事务已不可靠必须重构Consumer消费延迟突增但CPU、内存正常网络抖动导致Consumer与Broker长连接断开重连期间消息堆积netstat -anp | grep :10911 | grep ESTABLISHED | wc -l查看与Broker 10911端口的连接数启用channelCloseIdletrue并配置heartbeatBrokerInterval1000010秒心跳连接数监控比CPU监控更重要。我们曾因连接数从5000掉到4999导致1%的消息被Broker拒绝5.1 独家技巧用mqadmin命令快速定位90%的问题mqadmin是RocketMQ自带的运维神器但多数人只会用clusterList。我总结出五个必会命令查Topic详情mqadmin topicStatus -t topic_order_create -n ns1:9876关键看queueTotal总队列数、brokerOffsetBroker最新位点、consumerOffsetConsumer位点、lastUpdateTimestamp最后更新时间。如果lastUpdateTimestamp是1小时前的说明Broker已停止写入。查Consumer进度mqadmin consumerProgress -g group_order_create_prod -n ns1:9876输出是JSON用jq .messageModel看消费模式Clustering或Broadcastjq .jstack看线程堆栈。当jstack里出现大量WAITING说明Consumer卡住了。查Broker状态mqadmin brokerStatus -b broker-a:10911 -n ns1:9876关键看putTps写入TPS、getTps读取TPS、dispatchMaxBuffer缓冲区使用率。如果dispatchMaxBuffer95%立刻扩容Broker。强制重置Consumer位点mqadmin resetOffsetByTime -g group_order_create_prod -t topic_order_create -s 1653993600000 -n ns1:9876-s参数是毫秒时间戳设为30分钟前的时间可快速跳过积压消息。慎用必须先确认积压消息已无业务价值。导出消息内容mqadmin queryMsgById -i C0A8010200002A9F0000000000000001 -n ns1:9876-i是msgId从Consumer日志里复制。可直接看到消息体、Key、Tag比查数据库快10倍。5.2 血泪教训那些文档里不会写的“潜规则”不要相信brokerOffset - consumerOffset的差值这个差值只是数字不代表真实堆积。真实堆积量要看InSyncReplicas是否2。我们曾因差值显示100万紧急扩容Consumer结果发现是InSyncReplicas1Broker已拒绝写入扩容毫无意义。Consumer Group名长度不能超255字符RocketMQ内部用Group名做ZooKeeper路径超长会报org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException。我们曾用UUID做Group名结果超长失败。NameServer地址必须用IP不能用域名因为RocketMQ客户端DNS解析不支持SRV记录域名解析失败时不会fallback到备用地址直接连接失败。必须用192.168.1.10:9876;192.168.1.11:9876这种格式。消息Key不是用来查消息的而是用来追踪的msg.setKeys(order_123)后你不能用这个Key去查消息只能在mqadmin queryMsgByKey命令里用。业务系统查消息必须用订单号查DBKey只是辅助定位。consumeThreadMax不是越大越好当线程数超过CPU核数*2时线程上下文切换开销会抵消吞吐收益。我们实测过在32核机器上consumeThreadMax64时TPS最高设为128时TPS反而下降15%。我在IoT平台项目上线前用这套排查技巧提前发现了Consumer的ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT配置被注释掉了避免了上线后设备离线告警延迟47秒的事故。技术没有银弹只有把每个工具用到极致才能把风险挡在生产环境之外。
阅读完成 · 觉得有帮助？
