企业官网建设流程全解析

RocketMQ 顺序消息，是“4 把锁”实现的

RocketMQ 的顺序消息可以分两层回答：应用开发层怎么做，源码层为什么能保证顺序。

应用层保证发送有序和消费有序；源码层则用四把锁保证一个队列同时只被一个客户端、一个线程消费。

一、先区分分区有序和全局有序

顺序消息就是消息的消费顺序与生产顺序一致。比如订单的创建、支付、完成必须依次处理，但不同订单之间可以并行。

分区有序

同一个sharding key的消息进入同一个队列，队列内严格 FIFO。典型做法是以订单 ID 或用户 ID 作为分区键。

全局有序

所有消息进入同一个队列，按 FIFO 处理。全局有序可以理解为只有一个分区的分区有序，因此并发度和性能最低。

二、应用开发层怎么做

顺序消息需要三个阶段共同保证：顺序发送、顺序存储、顺序消费。RocketMQ 的 MessageQueue 天然保证 FIFO，因此应用层主要解决发送和消费两个问题。

发送有序

发送端用MessageQueueSelector把同一个业务键的消息固定路由到一个队列：

SendResultresult=producer.send(msg,newMessageQueueSelector(){@OverridepublicMessageQueueselect(List<MessageQueue>mqs,Messagemsg,Objectarg){LongorderId=(Long)arg;longindex=orderId%mqs.size();returnmqs.get((int)index);}},orderId);

建议先对业务键做 hash，再对队列数取模，减少数据倾斜。

消费有序

消费端不能使用默认的并发监听器，而要注册MessageListenerOrderly：

consumer.registerMessageListener(newMessageListenerOrderly(){@OverridepublicConsumeOrderlyStatusconsumeMessage(List<MessageExt>msgs,ConsumeOrderlyContextcontext){// 同一队列内串行处理returnConsumeOrderlyStatus.SUCCESS;}});

顺序消费失败时，RocketMQ 会不断重试，且不会跳过当前消息。如果一条消息始终失败，后续消息会被阻塞，所以要谨慎处理异常和积压。

三、源码层为什么需要四把锁

消费有序需要满足两个条件：

1. 一个分区只能投递给同一个客户端；
2. 一个客户端同一时间只能有一个线程消费同一个队列。

第一个条件用 Broker 端分布式锁实现，第二个条件用本地消费锁实现。再加上保护分布式锁操作的 Broker 全局锁，以及本地队列操作锁，一共四把锁。

四、第一把锁：Broker 端分布式锁

Broker 通过RebalanceLockManager管理队列锁：

ConcurrentMap<String,ConcurrentHashMap<MessageQueue,LockEntry>>mqLockTable;

这个结构保证同一个消费组下，同一个MessageQueue只会被一个消费者客户端锁定。

客户端锁定队列时，Broker 由AdminBrokerProcessor.lockBatchMQ()处理：

Set<MessageQueue>lockOKMQSet=this.brokerController.getRebalanceLockManager().tryLockBatch(requestBody.getConsumerGroup(),requestBody.getMqSet(),requestBody.getClientId());

tryLockBatch会遍历队列：如果锁已被自己持有，更新时间；如果锁已过期，则抢占；否则说明被其他消费者锁定，继续尝试下一个队列。

五、第二把锁：Broker 端全局锁

Broker 处理 RPC 请求的线程不止一个。如果多个线程同时修改mqLockTable，可能破坏锁分配的原子性。

所以在真正修改队列锁之前，会先进入 Broker 端的全局重入锁：

if(!notLockedMqs.isEmpty()){try{this.lock.lockInterruptibly();// 操作 this.mqLockTable}finally{this.lock.unlock();}}

这里的全局锁是保护“分布式锁本身”的，不能和队列锁混淆。

六、客户端怎么拿到分布式锁

RocketMQ 客户端启动后会运行多个异步服务：

• RebalanceService：负责负载均衡和再平衡；
• PullMessageService：负责拉取消息；
• ConsumeMessageOrderlyService：负责顺序消费。

Rebalance 完成后，客户端需要先向 Broker 申请队列锁。RebalanceImpl.lockAll()按 Broker 分组，发送同步锁队列请求：

Set<MessageQueue>lockOKMQSet=this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(),requestBody,1000);for(MessageQueuemq:lockOKMQSet){ProcessQueueprocessQueue=this.processQueueTable.get(mq);if(processQueue!=null){processQueue.setLocked(true);processQueue.setLastLockTimestamp(System.currentTimeMillis());}}

只有拿到锁的队列才会把本地ProcessQueue标记为已锁定。

七、消息拉取前先检查锁

pullMessage在拉取前会检查processQueue.isLocked()。如果本地快照没有锁，就延迟 3 秒再把PullRequest放回队列：

if(processQueue.isLocked()){if(!pullRequest.isLockedFirst()){finallongoffset=this.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());pullRequest.setLockedFirst(true);pullRequest.setNextOffset(offset);}}else{this.executePullRequestLater(pullRequest,pullTimeDelayMillsWhenException);return;}

这样能避免队列被多个客户端同时拉取。

八、第三、四把锁：本地消费锁

即使同一个队列只被一个客户端拉取，消费线程仍然可能并发处理。ConsumeMessageOrderlyService.run()中还有两级本地锁。

第一级是队列对象锁，锁住取消息和后续处理：

finalObjectobjLock=messageQueueLock.fetchLockObject(this.messageQueue);synchronized(objLock){List<MessageExt>msgs=this.processQueue.takeMessages(consumeBatchSize);// ...}

第二级是消费快照的ReentrantLock，只锁真正的业务消费：

try{this.processQueue.getLockConsume().lock();status=messageListener.consumeMessage(Collections.unmodifiableList(msgs),context);}finally{this.processQueue.getLockConsume().unlock();}

简单理解：

• 队列对象锁粒度更大，覆盖取消息、消费和位点提交；
• 消费锁粒度更小，只保证同一队列的业务消费只有一个线程执行。

九、总结

RocketMQ 顺序消息的四把锁分别是：

1. Broker 端分布式锁：一个队列只分配给一个客户端；
2. Broker 端全局锁：保护队列锁的原子操作；
3. 本地队列对象锁：串行化同一队列的消息获取和位点提交；
4. 本地消费锁：保证同一队列只有一个线程执行消费。

这套机制让同一个 MessageQueue 的数据最终被单个客户端、单线程串行消费。但消费者扩容、重启或 Broker 宕机时仍可能出现短时间乱序，所以业务层还要做好幂等。

另外要重点监控顺序消息的积压。因为顺序消费失败会持续重试，容易阻塞整个队列，发现问题要及时处理。