写这篇东西的起因是去年我们组一个订单系统莫名其妙发生消息积压业务方盯着监控问我“消费线程明明在跑怎么堆积就是不降”我排查了半天最后把问题定位在客户端消费进度上报和重平衡逻辑上。当时我就觉得很多项目用RocketMQ只是停留在调API层面真的出问题的时候对客户端内部机制的理解深度直接决定了你排查效率。所以这篇文章想干一件事把RocketMQ客户端消息机制从实战用法到源码实现彻彻底底串一遍。内容会覆盖Producer发送链路、Consumer消费链路、进度管理、重试死信、顺序消息和事务消息这几个核心模块全程基于我平时读源码的笔记和线上踩坑的经验不写空话只讲干货。这篇文章适合已经用RocketMQ做过一段时间开发、想进阶理解其内部原理的读者。如果你是刚接触消息中间件建议先把API跑通再看源码否则部分线程模型和协议细节会比较吃力。我后面所有源码分析都基于rocketmq 4.9.4版本整体结构和5.x经典模式差别不大看4.9.4反而是理解核心机制最好的起点。1. 先理清客户端消息机制的三条主线很多人读RocketMQ客户端源码觉得乱原因是不知道从哪下手。我自己的经验是先不要急着看类关系而是从消息的生命周期去切分。客户端消息机制本质上是三条并行且互相咬合的链路生产端发送链路、消费端拉取与消费链路、消费进度与可靠性保障链路。生产端发送链路从业务调用producer.send(msg)开始中间经历消息体组装、Topic路由查找、队列选择、网络发送、Broker存储确认最后把SendResult返回到业务线程。这条链路的终点在Broker但所有决策逻辑都在客户端完成包括发哪个队列、重试哪个Broker、同步还是异步。消费端拉取与消费链路从consumer.start()开始后台线程周期性向Broker拉取消息把消息放进本地缓存队列再由业务消费线程池取出并执行监听器。这条链路最复杂的地方在于Push模式其实是“客户端主动拉取 Broker长轮询挂起”的组合而不是Broker主动推送这个认知不建立起来后面看源码会很别扭。消费进度与可靠性链路解决的是“消息到底消费到哪了”和“消费失败的消息怎么办”两个问题。消费进度由客户端定时上报到Broker重试消息进入专门的重试Topic超过重试次数后落入死信队列。这条链路往往是最容易被忽视的但线上绝大部分诡异问题都出在这里。把这三条主线在脑子里立起来之后再去读源码就有地图了。建议的阅读顺序是先跑通一个生产者消费者Demo然后从DefaultMQProducerImpl.sendDefaultImpl入口追发送链路再从DefaultMQPushConsumer的启动逻辑和后台线程设计开始追消费链路。这篇文章就按这个顺序展开。2. 生产端发送机制从API到底层网络的完整路径2.1 三种发送方式的选型逻辑生产端发送方式有三种同步、异步、单向。我见过不少团队在同一套系统里只用同步发送理由是“稳妥”其实这是对异步的误解。选型应该根据业务对可靠性和时延的容忍度来定。发送方式返回值发送流程可靠性适用场景同步发送有SendResult发送后阻塞等待Broker确认高结果判断明确关键业务消息如订单支付通知异步发送回调函数发送后线程不阻塞结果在回调中获取高但错误处理在回调里高吞吐、对时延敏感如日志收集单向发送无返回发送后立即返回不关心结果低可能丢失日志、监控指标等允许丢失的场景源码里这三种方式都汇聚在DefaultMQProducerImpl中区别只在invokeSync、invokeAsync和invokeOneway三个调用。同步发送最核心的是拿到SendResult之后必须检查sendResult.getSendStatus()这个检查在源码层不会替你兜底。有些团队把send返回就当成功然后丢消息了才发现Broker返回了SLAVE_NOT_AVAILABLE或FLUSH_DISK_TIMEOUT。异步发送回调里必须处理Throwable因为网络异常、Broker主从切换时的失败都会通过异常抛到回调里。2.2 发送链路源码拆解从send到网络传输同步发送的入口方法是DefaultMQProducerImpl.sendDefaultImpl这是整个生产端最值得读的方法。一段一段拆开看一共做了四件事。第一步是消息格式校验和预处理。代码里会检查message.getTopic()是否为空、消息体长度是否超过maxMessageSize默认4MB然后调用MessageAccessor.putProperty给消息设置WAIT属性标记是否需要等待Broker存储确认。这一步看似简单但uniqID的生成逻辑就在附近。第二步是Topic路由查找。通过tryFindTopicBrokerList方法从本地缓存TopicRouteData中找一个Topic对应的Broker列表。如果缓存没有会调用fetchDefaultTopicRouteDataFromNameServer去NameServer拉取。这里有超时保护默认3秒如果NameServer都连不上直接抛MQClientException。所以生产端依赖NameServer的高可用这个环节要特别注意。第三步是关键——队列选择。正常情况下调用selectOneMessageQueue源码逻辑是先看sendLatencyFaultEnable开关默认关闭时走简单轮询内部维护一个ThreadLocalAtomicInteger每次发送对TopicPublishInfo中的队列数取模实现消息在多队列间轮流发送。开启故障延迟机制后会通过MQFaultStrategy维护一个延迟表优先避开最近发送失败的Broker不仅能切换队列还能切换Broker。这也是RocketMQ比Kafka在生产端高可用上做得更细的一个点。第四步是内核发送sendKernelImpl。这里做的事包括构造SendMessageRequestHeader包含ProducerGroup、Topic、QueueId、BornTimestamp等关键字段、调用NettyRemotingClient的invokeSync走网络层、最后把返回的SendResult包装后返回。整个调用链里还有一个容易被忽视的细节消息体压缩。当消息体大小超过4KB时会触发MessageCompressionUtil.compress压缩后会在消息属性里标记MESSAGE_COMPRESS_TYPE。这意味着消费端取消息时要判断压缩标记并解压好在4.x的DefaultMQPushConsumer已经自动处理了。这个发送链路里同步和异步在sendKernelImpl里分道扬镳。异步发送走invokeAsync时会先进入一个Semaphore信号量控制默认useSemaphoreAsynctrue最多允许65535个异步请求在途。如果超过这个量请求会被直接丢弃并返回RejectSendRequestException。所以异步发送的流量控制是客户端自己实现的不是靠Broker限流。2.3 唯一ID与消息去重的底层依据很多人在消费端做幂等时会用到消息的ID但多数人分不清楚msgId和uniqID的区别。源码层面uniqID是客户端在发送前由MessageClientIDSetter.createUniqID生成的本质上是一个定长的字节数组里面编码了进程IP、进程PID、类加载时间戳和一个AtomicInteger自增序号。而msgId是Broker收到消息后生成并返回的存储在消息的MSG_ID属性中格式是Broker地址加时间戳加序号。这两个ID都存在于消息体里消费端通过message.getMsgId()拿到的是Broker返回的msgId。做幂等设计时我推荐使用uniqID因为它是客户端全局唯一的业务语义ID而msgId依赖Broker重启后时间戳的解析虽然唯一性也没问题但用于跨系统追踪时语义不如uniqID清晰。更稳妥的做法是业务自己生成一个业务唯一键放进消息的keys字段或属性里消费端拿这个键做幂等这比依赖RocketMQ的ID更可控。2.4 生产端实战的常见坑发送超时时间。send(msg, timeout)默认是3秒这个值在正常内网环境够用但如果业务消息体大、或Broker磁盘IO抖动3秒很容易触发RemotingTimeoutException。我见过有人为了追求速度把超时降到1秒结果高峰期大量发送失败反而拖垮了业务。建议保留默认或调大到5秒同时做好失败重试。批量发送大小限制。sendBatch方法底层逐条发送并没有真正在协议层合并为批量存储。批量API要求所有消息同一Topic、同一等待存储确认策略且总大小不要超过1MB。超过1MB时Broker会拒收这个限制很多人不知道。异步回调线程模型。异步发送的回调是在Netty的EventExecutor线程中执行的不是业务线程。所以回调里不能做耗时操作否则会阻塞Netty的IO线程影响其他消息的发送。正确的做法是回调里只做状态记录把耗时的业务逻辑丢给独立的线程池。NameServer不可用与生产端可用性之间的缓冲。生产端在启动时会从NameServer拉取Topic路由然后缓存到本地。如果运行期间NameServer全部宕机已经建好的连接还能继续发送吗答案是一段时间内能。因为MQClientInstance在拉取不到路由时会继续使用本地缓存但一旦本地缓存的Broker状态失效比如Broker宕机又没有新的路由来源发送就会失败。所以生产端高可用不能只靠RocketMQ自身NameServer的可用性必须纳入监控。3. 消费端机制Push模型背后的引擎与灵魂3.1 先纠正一个概念Push其实是Pull很多文章把RocketMQ的PushConsumer理解为“消息推送”这是完全错误的。DefaultMQPushConsumer虽然名字里有Push但底层的PullMessageService线程是主动去向Broker拉消息的。所谓Push是Broker收到拉取请求后如果没有新消息不会立刻返回空响应而是把请求挂起长轮询默认挂起15秒等有新消息到达时再返回。这个机制叫longPolling实际上就是服务端hold住客户端请求让客户端看起来像“消息来了”。所以RocketMQ的Push模式 客户端主动Pull Broker长轮询挂住请求。理解这一点你就明白为什么消费端拉取频率并不高却能做到接近实时的消费延迟。Kafka的老版本没有长轮询时消费者只能靠频繁pull来保证低延迟但代价是大量无效请求。RocketMQ的长轮询设计在这一点上更优雅。3.2 DefaultMQPushConsumer的启动逻辑与两个引擎线程DefaultMQPushConsumer.start()做的事情排个序初始化MQClientInstance、注册消费者到BrokerrebalanceImmediately、启动RebalanceService线程、启动PullMessageService线程、启动消息消费服务。这里最值得研究的是两个后台线程RebalanceService和PullMessageService。RebalanceService每隔20秒执行一次重平衡逻辑检查当前ConsumerGroup下的消费者列表是否发生变化。变化的来源是Broker上的消费者注册信息每个消费者启动时会把自己的clientId注册到BrokerBroker维护一个消费者列表。如果新增了一个消费者实例或某个旧实例宕机了心跳超时被移除RebalanceService就要重新分配每个消费者负责哪些消息队列。PullMessageService则是一个无限循环线程。它的做法简单但有效每次循环检查所有消费者队列是否存在待拉取的请求有就调用pullMessage走网络请求。不要觉得它很傻——它内部有一个LinkedBlockingQueue作为拉取任务的阻塞队列PullRequest会提前被准备好哪个队列该拉了、该拉多少条都由之前的拉取结果和消费速度决定。这两个线程的协作关系是我觉得整个消费端最精妙的地方。RebalanceService负责回答“每个消费者应该处理哪些队列”PullMessageService负责回答“这些队列的消息怎么拉回来”。两者之间通过一个pullRequestQueue解耦重平衡不直接打断拉取流程而是通过更新ProcessQueue状态来间接影响。3.3 重平衡机制与队列分配策略重平衡的核心方法是RebalanceImpl.rebalanceByTopic。它的流程是从Broker获取当前Group下的消费者ID列表结合Topic下的队列列表调用分配策略默认AllocateMessageQueueAveragely计算本消费者应处理的队列集合然后对比当前正在处理的队列集合该释放的释放、该占用的占用。这里的分配策略非常重要因为它的选择直接影响负载是否均匀。AllocateMessageQueueAveragely是把队列按消费者数量平均分不保证按消费者索引连续分配比如3个队列分给2个消费者可能是消费者A拿2个、消费者B拿1个。官方还有AllocateMessageQueueByMachineRoom、AllocateMessageQueueConsistentHash等策略但生产环境用默认的平均分配就够了。重平衡触发时最大的坑是“消费中消息被分配到其他消费者”。比如一个消费者正在消费某个队列的消息此时Group里新加入一个实例重平衡把这个队列分配给新实例。旧的消费者其实还持有ProcessQueue里的本地缓存消息这些消息会被直接丢弃新消费者会从Broker记录的消费进度处继续拉取。如果消费进度还没有来得及上报就会出现重复消费。这个机制是重复消息的一个重要来源后面讲幂等时你会看得更清楚。3.4 ProcessQueue本地缓冲队列的灵魂ProcessQueue是消费端最核心的类它不只是“本地缓存”那么简单。每个消息队列在消费端对应一个ProcessQueue它维护了三个关键状态消息缓存一个TreeMapLong, MessageExtkey是消息在队列中的offset、正在消费的窗口、消费进度快照。拉取消息时PullMessageService把从Broker拿到的消息放进对应ProcessQueue的缓存中同时记录nextPullOffset。消费线程池从ProcessQueue的缓存里取出消息执行监听器成功就从这个队列移除并更新消费进度。如果消费失败就进入重试流程而ProcessQueue里的消息不会被移除。ProcessQueue还承担了消费端限流的功能。源码里pullMessage方法会检查ProcessQueue当前缓存的消息数量和大小如果超过阈值默认pullThresholdForQueue为1000条pullThresholdSizeForQueue为100MB就不会再发起新的拉取直到消费线程把缓存消耗下去。这就是RocketMQ消费端背压机制的实现没有这一层Broker发多少消费者就存多少内存早就爆了。ProcessQueue的状态字段dropped也很关键。当重平衡把这个队列分给别的消费者时旧的ProcessQueue会被标记为droppedtrue这样即使还有消息在本地缓存里消费线程也会停止处理然后丢弃。这就是前面提到的消费中消息丢失的根源之一。3.5 并发消费与顺序消费的线程模型差异DefaultMQPushConsumer注册的MessageListener有两种实现MessageListenerConcurrently和MessageListenerOrderly。这两种监听器对应完全不同的消费服务。并发消费走ConsumeMessageConcurrentlyService。默认线程池是ThreadPoolExecutor核心线程数和最大线程数都等于consumeThreadMin默认20。每一条消息提交成独立任务丢给线程池多个线程并行消费同一个ProcessQueue里的消息它们之间没有顺序保证。这也是为什么并发模式下的消费效率高但需要业务侧自己保证幂等。顺序消费走ConsumeMessageOrderlyService。这里它的线程池是单线程的每个队列的消息由同一个线程串行执行。真正的顺序保证不仅依赖Consumer的单线程执行还要配合队列锁机制。ConsumeMessageOrderlyService在消费前会尝试获取ProcessQueue对应的队列锁源码里的lockMap拿到锁才允许从缓存里取消息执行。这就防止了重平衡期间新旧消费者同时消费同一个队列导致乱序。所以顺序消息是“三端配合”Producer端通过MessageQueueSelector保证同一业务消息进同一队列Broker端保证同一队列消息顺序存储Consumer端用单线程队列锁保证串行消费。任何一端没配好顺序就保证不了。我见过有人在生产上只用producer.send(msg)不指定selector然后消费端用了顺序监听器结果消息依然乱序就是因为同一个业务的多个消息落到了不同队列里。3.6 消费进度的管理与上报机制消费进度是消息中间件可靠性的最后一道防线。RocketMQ里面有两种进度存储实现RemoteBrokerOffsetStore和LocalFileOffsetStore。Push模式默认使用RemoteBrokerOffsetStore。消费线程每消费完一条消息会调用updateOffset更新内存中的offset后台有个定时任务默认5秒把内存中的offset批量上报到BrokerBroker把它持久化到内部TopicCONSUMER_OFFSET里。这样一旦消费者重启或发生重平衡新消费者就从Broker记录的offset处开始拉取。Pull模式默认使用LocalFileOffsetStore进度存储在本地的offset文件里路径在用户目录下/store/config/consumerOffset.json。进度上报时机这里有个细节上报不是实时的是定时批量上报。如果消费者在处理完一条消息后立刻宕机且5秒上报周期没到这条消息会在重启后被重新拉取造成重复消费。这是重复消息的另一个根源而且几乎无法完全避免。所以做消费端设计时必须把“至少一次”的投递语义当成默认前提。订阅关系是进度管理里的隐藏坑。subscribe(topic, tag)里的tag源码底层会转换为一个正则表达式过滤器比如TagA || TagC。如果同一个ConsumerGroup下的多个消费者订阅的tag不一致重平衡时后加入的消费者拿到队列后会按照自己的过滤器拉取消息会导致部分消息“丢失”实际上是被过滤掉了。RocketMQ官方明确要求同一Group下的订阅关系必须一致这个约束在源码注释里写了但很多人不看。4. 可靠性机制重试、死信与消费幂等4.1 消费失败后的重试链路消费端返回RECONSUME_LATER或抛出异常后消息并不会简单地被丢弃而是进入一条设计精密的延迟重试链路。我先说流程再讲源码。流程是这样的Broker收到消费失败的消息后会把消息转入一个特殊Topic叫%RETRY% ConsumerGroup名。这个Topic的queueCount是1所以叫重试队列。每条重试消息被投递回来之前会在原消息属性里设置一个延迟级别——第一次重试延迟1秒第二次5秒第三次10秒往后依次递增直到第16次延迟2小时后如果还消费失败消息会被转入死信队列%DLQ% ConsumerGroup名。这个延迟级别对应Broker的messageDelayLevel配置默认有18个级别1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。重试消息在Broker端不是真的等上1秒再投递而是消息先写入CommitLog然后被放入一个延迟队列ScheduleMessageService由后台线程扫描到期消息到期后重新投递给客户端。源码里和这个机制对接的消费端逻辑在ConsumeMessageConcurrentlyService.processConsumeResult方法里。这个方法拿到消费者返回的结果然后调用submitConsumeRequestLater重新构建一个消费请求或者使用sendMessageBack把消息送回Broker。调用sendMessageBack时源码会计算一个重试次数delayLevel传给Broker。你如果直接看消息体属性会看到一个叫RETRY_TOPIC的属性里面存放的就是原始Topic名。消费端拉取消息时发现RETRY_TOPIC属性存在就会先从重试队列解析出原始Topic信息再投递给业务监听器。4.2 死信队列的创建与消费超过最大重试次数默认16次的消息自动进入%DLQ% ConsumerGroup名的死信Topic。死信队列默认queueCount是1。我在生产上有个习惯专门写一个死信消费者监听%DLQ%的Topic把最新一条死信的完整字段消息体、keys、reconsumeTimes记到一个告警表里用于事后人工排查。死信队列有一个反直觉的点——它也是可以被消费者消费的。如果你不想让消息静默死掉可以主动订阅%DLQ% 自己的Group名消费端逻辑就变得非常灵活可以自动把死信消息转发存库也可以写告警日志。但注意死信队列里的消息不会自动恢复回原Topic需要你自己处理。重试次数属性reconsumeTimes在消息里是递增的Broker根据它判断重试是否结束。消费端入参MessageExt里有现成的getReconsumeTimes()方法业务消费时如果发现超过N次就可以放弃该消息避免无意义的反复重试。4.3 消息重复的四个主要来源与幂等设计说一个让很多人震惊的事实RocketMQ本身不提供“不重复消费”的语义保证它提供的是“至少一次at least once”语义。也就是说一条消息有可能在异常情况下被消费多次。重复的来源主要有四个我列个清单方便排查。第一消费成功后进度未来得及上报就宕机。前面说过进度是5秒定时上报宕机窗口内必然重复。第二重平衡导致消息队列转移旧消费实例的ProcessQueue被丢弃但消息已被处理过。第三消费超时后的重新投递。Broker侧有专门的机制如果一条消息在消费处理中长时间未返回且连接状态异常Broker会认为该消息未被消费然后重新投递。第四业务代码自身导致的重试比如监听器里调外部接口超时外层框架把异常吞掉返回RECONSUME_LATER但其实消息已经被本地处理过。既然重复无法从中间件层面消除那消费端幂等就是唯一出路。我实践下来的最佳做法是设计一张去重表或使用Redis的SETNX把业务唯一键作为去重条件。去重键建议优先用消息属性里的自定义业务键在发送时通过message.setKeys(orderId)设置其次用uniqID。查库去重和Redis去重各有优劣数据量小的直接查库最稳。4.4 消费端限流与推送背压消费端的限流是很多人忽视的部分。虽然ProcessQueue有缓存阈值会暂停拉取但如果你并发消费线程池设置得过大会出现一个尴尬的问题线程全在等IO消息全躺在内存里Broker消息消费速度看着很快实际上很多消息还没被业务处理完。我见过有团队把consumeThreadMax调到64结果堆内存直接被打满消费线程全部阻塞在DB查询上。正确的调优思路是根据下游的消费能力反向设置消费线程数。如果消费逻辑只是简单写缓存consumeThreadMin20够用如果消费逻辑涉及查库、调接口、写文件等IO密集操作线程数不要超过下游能承受的并发数。假如下游接口的QPS上限是500消费线程再大也只会增加无意义的重试和超时。ConsumeMessageConcurrentlyService里还有个参数consumeMessageBatchMaxSize默认1它控制一次拉取的消息最多有多少条会放进同一个消费任务。批量消费模式下同一个ConsumeRequest里可能包含多条消息监听器拿到的ListMessageExt就是这批消息。批量消费能减少线程调度开销但如果消费者内部处理时有一个消息处理失败整批消息都会返回RECONSUME_LATER导致同批里其他消息被重复消费。所以我一般建议consumeMessageBatchMaxSize不要调得太大1~5之间比较合适。5. 特殊消息机制的实战与源码顺序消息和事务消息5.1 顺序消息为什么必须三端配合顺序消息是RocketMQ最有价值的特性之一但也是实现起来最容易出错的一个。前面讲消费端线程模型时已经提到过三端配合这里再展开说说为什么必须这样设计。Producer端必须使用send(msg, new MessageQueueSelector() {...})这个重载。MessageQueueSelector接口的select方法会返回一个队列ID源码里DefaultMQProducerImpl.sendSelectImpl拿到这个返回值后会把消息发送到指定的队列不会走默认的轮询逻辑。如果业务想在全局范围内保证顺序比如同一个订单ID的所有状态变更消息必须顺序消费那select方法里就基于订单ID做hash让同一个订单ID的消息始终落入同一个队列。Broker端对同一个队列内消息的写入和存储天然是顺序的因为底层是单一CommitLog顺序追加。这里有一个前提——如果不使用事务消息单条消息的发送是原子的所以队列内无并发写入冲突。Consumer端必须使用MessageListenerOrderly监听器。这个监听器绑定的ConsumeMessageOrderlyService在消费前会执行lockMessageQueue操作获取队列锁保证同一队列同一时刻只有一个消费者在处理。配合单线程ConsumeRequest消息就能按队列内顺序被串行消费。同时在消费成功的commit过程中只有前一条消息处理完成后下一条消息才会进入业务监听器。乱序问题绝大多数出现在Producer端队列选择上。我见过一个线上案例团队用了顺序消息但select方法是从一个Map里取队列IDMap的key是随机生成的字符串结果同一个订单ID的消息分散到两个队列里消费端再怎么加锁也没用。排查方式就是看Broker端消息的queueId分布如果同业务ID的消息出现在不同队列直接锁定KPI。5.2 事务消息的Half Message机制事务消息解决的是“本地事务与消息发送的最终一致性”问题。场景比如下单业务先写订单数据库再发一条订单消息。如果先写库再发消息消息发送失败就会造成订单已建但通知缺失如果先发消息再写库消费端可能看到不存在的订单。RocketMQ用Half Message机制把这两个动作原子化了。源码链路是这样的TransactionMQProducer.sendMessageInTransaction方法会调用DefaultMQProducerImpl的下层逻辑发送前的核心在AsyncRunnable线程执行它会先构造一条消息把业务Topic存在消息属性里然后把消息的Topic地址替换成RMQ_SYS_TRANS_HALF_TOPIC也就是半消息队列。半消息是特殊的它对普通消费者不可见专门用于事务流程跟踪。半消息发送成功后客户端立刻执行TransactionListener.executeLocalTransaction回调在这个回调里执行本地事务逻辑写订单库。本地事务执行完成后客户端调用endTransaction向Broker发送提交或回滚指令。Broker收到commit指令后会把消息从半消息队列恢复到真正的业务Topic中此时消费者才可见。这里有个非常重要的细节如果本地事务执行时进程挂了endTransaction永远不会发送半消息就会一直停留在半消息队列中。RocketMQ的应对措施是Broker端的定时扫描器它每隔一段时间扫描半消息队列中停留超过checkTransactionTimeout的消息然后向Producer客户端发起回查请求调用TransactionListener.checkLocalTransaction确认本地事务的结果再决定commit或rollback。所以事务消息的最终一致性是靠“半消息 回查机制”两段设计协同完成的。实现TransactionListener时executeLocalTransaction方法里建议把事务执行结果先持久化下来比如记录在业务表的状态字段里这样回查时checkLocalTransaction才能准确判断事务到底成功还是失败。如果回查时发现本地事务还没执行完返回UNKNOWN让Broker继续等待并再次回查直到明确成功或失败。5.3 事务消息的实战注意点事务消息不是万能的它的代价是额外的消息写入和回查开销。半消息本身也会占用CommitLog空间回查频率过高会加重Broker负担。所以不建议把每个普通业务都包成事务消息只建议用于“必须保证本地操作与消息发送一致的场景”。事务消息不支持批量发送。sendMessageInTransaction走的是一个特殊的单条发送链路批量接口在事务模式下不可用。事务消息的消费端没有任何特殊控制。当Broker把事务消息恢复到业务Topic后消费端的表现和普通消息完全一样所以消费端仍然需要幂等设计。事务消息解决的是生产端的原子性问题消费端的语义依然是“至少一次”。6. 踩坑实录客户端消息机制的线上问题排查6.1 消息堆积不消费但消费线程池很忙碌这是一个很典型的现象监控面板上看消费者Group的堆积数量持续上涨消费线程池的活跃线程数也不低但消费总量没有突破下限。这种问题的根源大概率是消费逻辑本身卡在某些慢操作上——比如每条消息都调外部接口外部接口响应慢线程全部等待在IO上没有新进展。源码层面看ConsumeMessageConcurrentlyService的线程池空闲时不会自动扩容因为核心线程数等于最大线程数。所以即使堆积再多线程数也不会增加。排查思路先看消费线程的栈日志jstack如果发现大量线程阻塞在同一个第三方调用上基本锁定问题之后要么优化消费逻辑缓存化要么拆成两步消费一个消费线程快速写本地队列后台批量处理要么根据下游能力调整consumeThreadMin与consumeMessageBatchMaxSize。6.2 重平衡期间消费重复量突增如果你发现消息重复量在某个时间点突增第一反应应该看那个时间点有没有消费者上线或下线。重平衡导致ProcessQueue被标记dropped进而丢弃未消费消息以及新消费者从旧offset重新拉取这两个机制叠加起来必然产生重复。源码里看RebalanceService每20秒扫描一次所以从消费者注册到Broker到触发重平衡存在一个窗口期。针对这个窗口期的重复只能靠消费端幂等兜底。如果你的业务绝对无法容忍重平衡期间的重复一个思路是把消费进度上报周期缩短到1秒对Broker压力很小同时消费完成后立刻调用updateOffset。只能说降低概率不能说完全消除。6.3 顺序消息莫名乱序顺序消息乱序的排查优先级如下。第一确认Producer端使用了MessageQueueSelector且选择逻辑基于业务键第二确认消费端使用的是MessageListenerOrderly第三检查重平衡期间是否发生过队列转移转移后新消费者从Broker进度开始拉取这就天然乱序了因为旧消费者已经处理了部分消息但进度没上报。最后一点只能缩小重平衡窗口或用幂等覆盖。还有一个容易忽略的坑在顺序消费模式里如果监听器返回SUSPEND_CURRENT_QUEUE_A_MOMENT顺序消费的失败语义源码会暂停当前队列的消费一段时间然后重新消费当前这条消息。在这个暂停期间队列锁还是被当前消费者持有的。如果这个消费者的线程长时间暂停重平衡服务会等队列锁超时后强制把队列分配给其他消费者这也是乱序的一个隐藏入口。6.4 发送时消息丢失无感知单向发送模式下消息丢失是无感知的这个前面说过。还有一个隐蔽场景异步发送的回调里没有处理异常发送失败后什么都不做看起来业务继续跑实际上消息丢了。排查时翻日志找不到任何错误因为回调里吞掉了异常。正确的做法异步回调里如果sendResult为null或status不是OK必须走补偿链路比如路由到备用Topic或直接记录到告警表。同时发送前生成业务唯一键放入消息属性发送失败后这个键还能用来查Broker底表或日志。6.5 消费端订阅关系不一致引发的“消息丢失”订阅关系一致性这条坑值得单独再说一次。源码里SubscriptionData的生成是每个消费者独立执行的重平衡时Broker会根据每个消费者上报的订阅信息做过滤。如果同一Group下实例A订阅TagA||TagB实例B订阅TagA两个实例的过滤条件不一致可能导致一部分消息在Broker做tag过滤时就被丢弃了。由于客户端默认开启tag过滤postSubscriptionWhenPull为false时自动生效这个问题隐蔽性极高。真正排查这类问题看监控里消息量在某个消费者上线后骤降同时堆积并没有良性上涨。此时立刻检查所有Consumer实例的订阅表达式统一改成一致的Tag列表。如果你还需兼容不同消费逻辑正确的做法是拆分成多个ConsumerGroup而不是在一个Group里用不同Tag。6.6 读源码的收益与建议说实话最初让我下决心啃客户端源码的契机是线上一个老问题反复出现消费端重启后总有一部分消息重复团队内部对“RocketMQ到底能不能保证不重复”争论不休。读完ProcessQueue和消费进度上报逻辑后所有争论都停下来了——源码白纸黑字写着5秒上报、本地缓存先删后报重复机制一目了然。我的建议是搭建一个Demo环境给消息打好日志时间戳用两三个消费者实例做重平衡测试亲眼看看消息的重复和延迟。源码配合实验比任何文档都直观。RocketMQ 5.x的客户端引入了Proxy模式和POP消费但核心的消息机制——队列模型、重平衡、进度上报、长轮询——依然是这套逻辑的内核理解4.9.4版本再去看5.x变化基本上没有障碍。最后分享一个小技巧读源码时先给类分一下组配套类比如TopicPublishInfo与TopicRouteData不用死磕核心链路类比如DefaultMQProducerImpl、DefaultMQPushConsumer、ProcessQueue、RebalanceImpl值得反复读。前三遍看不懂没关系第四遍开始你就知道哪些方法是核心、哪些字段是状态标记了。
