RocketMQ是一款 分布式、队列模型的消息中间件，由阿里巴巴团队研发，借鉴参考了JMS规范的MQ实现，更参考了优秀的开源消息中间件KAFKA，并结合阿里实际业务需求，在天猫双十一的场景，实现业务消峰、分布式事物的优秀框架。

其具有以下特点：

1.能够保证严格的消息顺序

2.提供丰富的消息拉取模式

3.高效的订阅者水平扩展能力

4.实时的消息订阅机制

5.亿级消息堆积能力

RocketMQ的顺序消息需要满足2点：

1.Producer端保证发送消息有序，且发送到同一个队列。

2.consumer端保证消费同一个队列。

如何在集群消费时保证消费的有序呢？

1.ConsumeMessageOrderlyService类的start()方法，如果是集群消费，则启动定时任务，定时向broker发送批量锁住当前正在消费的队列集合的消息，具体是consumer端拿到正在消费的队列集合，发送锁住队列的消息至broker，broker端返回锁住成功的队列集合。consumer收到后，设置是否锁住标志位。

这里注意2个变量：

consumer端的RebalanceImpl里的ConcurrentHashMap

processQueueTable，是否锁住设置在ProcessQueue里。

broker端的RebalanceLockManager里的ConcurrentHashMap mqLockTable，这里维护着全局队列锁。

2.ConsumeMessageOrderlyService.ConsumeRequest的run方法是消费消息，这里还有个MessageQueueLock messageQueueLock，维护当前consumer端的本地队列锁。保证当前只有一个线程能够进行消费。

3.拉到消息存入ProcessQueue，然后判断，本地是否获得锁，全局队列是否被锁住，然后从ProcessQueue里取出消息，用MessageListenerOrderly进行消费。拉到消息后调用ProcessQueue.putMessage(final List msgs)存入，具体是存入TreeMap msgTreeMap。然后是调用ProcessQueue.takeMessags(final

int batchSize)消费，具体是把msgTreeMap里消费过的消息，转移到TreeMap msgTreeMapTemp。

4.本地消费的事务控制，ConsumeOrderlyStatus.SUCCESS（提交），ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT（挂起一会再消费），在此之前还有一个变量ConsumeOrderlyContext context的setAutoCommit()是否自动提交。

当SUSPEND_CURRENT_QUEUE_A_MOMENT时，autoCommit设置为true或者false没有区别，本质跟消费相反，把消息从msgTreeMapTemp转移回msgTreeMap，等待下次消费。

当SUCCESS时，autoCommit设置为true时比设置为false多做了2个动作，consumeRequest.getProcessQueue().commit()和this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(),commitOffset, false);

ProcessQueue.commit()：本质是删除msgTreeMapTemp里的消息，msgTreeMapTemp里的消息在上面消费时从msgTreeMap转移过来的。

this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset()：本质是把拉消息的偏移量更新到本地内存中，然后定时更新到broker。

那么少了这2个动作会怎么样呢，随着消息的消费进行，msgTreeMapTemp里的消息堆积越来越多，消费消息的偏移量一直没有更新到broker导致consumer每次重新启动后都要从头开始重复消费。就算更新了offset到broker，那么msgTreeMapTemp里的消息堆积呢？不知道这算不算bug所以，还是把autoCommit设置为true吧。

最后要感谢这个优秀的平台，可以让我们相互交流，如果想进一步学习交流，可以加群460570824，希望大家可以一起学习进步！