本文首先重现网友提出的问题，然后对其进行原理分析，然后验证猜想，并给出实战建议。
抛出问题
一个新的消费组订阅一个已存在的Topic主题时，消费组是从该Topic的哪条消息开始消费呢？
首先翻阅DefaultMQPushConsumer的API时，setConsumeFromWhere(ConsumeFromWhere consumeFromWhere)API映入眼帘，从字面意思来看是设置消费者从哪里开始消费，正是解开该问题的”钥匙“。
ConsumeFromWhere枚举类图如下：
- CONSUME_FROM_MAX_OFFSET
从消费队列最大的偏移量开始消费。 - CONSUME_FROM_FIRST_OFFSET
从消费队列最小偏移量开始消费。 - CONSUME_FROM_TIMESTAMP
从指定的时间戳开始消费，默认为消费者启动之前的30分钟处开始消费。可以通过DefaultMQPushConsumer#setConsumeTimestamp。
是不是点小激动，还不快试试。
需求：新的消费组启动时，从队列最后开始消费，即只消费启动后发送到消息服务器后的最新消息。
环境准备
本示例所用到的Topic路由信息如下：
Broker的配置如下(broker.conf)
1brokerClusterName = DefaultCluster 2brokerName = broker-a 3brokerId = 0 4deleteWhen = 04 5fileReservedTime = 48 6brokerRole = ASYNC_MASTER 7flushDiskType = ASYNC_FLUSH 8 9storePathRootDir=E:/SH2019/tmp/rocketmq_home/rocketmq4.5_simple/store 10storePathCommitLog=E:/SH2019/tmp/rocketmq_home/rocketmq4.5_simple/store/commitlog 11namesrvAddr=127.0.0.1:9876 12autoCreateTopicEnable=false 13mapedFileSizeCommitLog=10240 14mapedFileSizeConsumeQueue=2000
其中重点修改了如下两个参数：
- mapedFileSizeCommitLog
单个commitlog文件的大小，这里使用10M，方便测试用。 - mapedFileSizeConsumeQueue
单个consumequeue队列长度，这里使用1000，表示一个consumequeue文件中包含1000个条目。
消息发送者代码
1public static void main(String[] args) throws MQClientException, InterruptedException { 2 DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name"); 3 producer.setNamesrvAddr("127.0.0.1:9876"); 4 producer.start(); 5 for (int i = 0; i < 300; i++) { 6 try { 7 Message msg = new Message("TopicTest" ,"TagA" , ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); 8 SendResult sendResult = producer.send(msg); 9 System.out.printf("%s%n", sendResult); 10 } catch (Exception e) { 11 e.printStackTrace(); 12 Thread.sleep(1000); 13 } 14 } 15 producer.shutdown(); 16}
通过上述，往TopicTest发送300条消息，发送完毕后，RocketMQ Broker存储结构如下：
消费端验证代码
1public static void main(String[] args) throws InterruptedException, MQClientException { 2 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_01"); 3 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); 4 consumer.subscribe("TopicTest", "*"); 5 consumer.setNamesrvAddr("127.0.0.1:9876"); 6 consumer.registerMessageListener(new MessageListenerConcurrently() { 7 @Override 8 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 9 ConsumeConcurrentlyContext context) { 10 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); 11 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 12 } 13 }); 14 consumer.start(); 15 System.out.printf("Consumer Started.%n"); 16}
执行上述代码后，按照期望，应该是不会消费任何消息，只有等生产者再发送消息后，才会对消息进行消费，事实是这样吗？执行效果如图所示：
令人意外的是，竟然从队列的最小偏移量开始消费了，这就“尴尬”了。难不成是RocketMQ的Bug。带着这个疑问，从源码的角度尝试来解读该问题，并指导我们实践。
探究CONSUME_FROM_MAX_OFFSET实现原理
对于一个新的消费组，无论是集群模式还是广播模式都不会存储该消费组的消费进度，可以理解为-1,此时就需要根据DefaultMQPushConsumer#consumeFromWhere属性来决定其从何处开始消费，首先我们需要找到其对应的处理入口。我们知道，消息消费者从Broker服务器拉取消息时，需要进行消费队列的负载，即RebalanceImpl。
温馨提示：本文不会详细介绍RocketMQ消息队列负载、消息拉取、消息消费逻辑，只会展示出通往该问题的简短流程，如想详细了解消息消费具体细节，建议购买笔者出版的《RocketMQ技术内幕》书籍。
RebalancePushImpl#computePullFromWhere
1public long computePullFromWhere(MessageQueue mq) { 2 long result = -1; // @1 3 final ConsumeFromWhere consumeFromWhere = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere(); 4 final OffsetStore offsetStore = this.defaultMQPushConsumerImpl.getOffsetStore(); 5 switch (consumeFromWhere) { 6 case CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST: 7 case CONSUME_FROM_MIN_OFFSET: 8 case CONSUME_FROM_MAX_OFFSET: 9 case CONSUME_FROM_LAST_OFFSET: { // @2 10 // 省略部分代码 11 break; 12 } 13 case CONSUME_FROM_FIRST_OFFSET: { // @3 14 // 省略部分代码 15 break; 16 } 17 case CONSUME_FROM_TIMESTAMP: { //@4 18 // 省略部分代码 19 break; 20 } 21 default: 22 break; 23 } 24 return result; // @5 25 }
代码@1：先解释几个局部变量。
- result
最终的返回结果，默认为-1。 - consumeFromWhere
消息消费者开始消费的策略，即CONSUME_FROM_LAST_OFFSET等。 - offsetStore
offset存储器，消费组消息偏移量存储实现器。
代码@2：CONSUME_FROM_LAST_OFFSET(从队列的最大偏移量开始消费)的处理逻辑，下文会详细介绍。
代码@3：CONSUME_FROM_FIRST_OFFSET(从队列最小偏移量开始消费)的处理逻辑，下文会详细介绍。
代码@4：CONSUME_FROM_TIMESTAMP(从指定时间戳开始消费)的处理逻辑，下文会详细介绍。
代码@5：返回最后计算的偏移量，从该偏移量出开始消费。
CONSUME_FROM_LAST_OFFSET计算逻辑
1case CONSUME_FROM_LAST_OFFSET: { 2 long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE); // @1 3 if (lastOffset >= 0) { // @2 4 result = lastOffset; 5 } 6 // First start,no offset 7 else if (-1 == lastOffset) { // @3 8 if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { 9 result = 0L; 10 } else { 11 try { 12 result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq); 13 } catch (MQClientException e) { // @4 14 result = -1; 15 } 16 } 17 } else { 18 result = -1; 19 } 20 break; 21}
代码@1：使用offsetStore从消息消费进度文件中读取消费消费进度，本文将以集群模式为例展开。稍后详细分析。
代码@2：如果返回的偏移量大于等于0，则直接使用该offset，这个也能理解，大于等于0，表示查询到有效的消息消费进度，从该有效进度开始消费，但我们要特别留意lastOffset为0是什么场景，因为返回0，并不会执行CONSUME_FROM_LAST_OFFSET(语义)。
代码@3：如果lastOffset为-1,表示当前并未存储其有效偏移量，可以理解为第一次消费，如果是消费组重试主题，从重试队列偏移量为0开始消费；如果是普通主题，则从队列当前的最大的有效偏移量开始消费，即CONSUME_FROM_LAST_OFFSET语义的实现。
代码@4：如果从远程服务拉取最大偏移量拉取异常或其他情况，则使用-1作为第一次拉取偏移量。
分析，上述执行的现象，虽然设置的是CONSUME_FROM_LAST_OFFSET，但现象是从队列的第一条消息开始消费，根据上述源码的分析，只有从消费组消费进度存储文件中取到的消息偏移量为0时，才会从第一条消息开始消费，故接下来重点分析消息消费进度存储器(OffsetStore)在什么情况下会返回0。
接下来我们将以集群模式来查看一下消息消费进度的查询逻辑，集群模式的消息进度存储管理器实现为：RemoteBrokerOffsetStore,最终Broker端的命令处理类为：ConsumerManageProcessor。
1ConsumerManageProcessor#queryConsumerOffset 2private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { 3 final RemotingCommand response = 4 RemotingCommand.createResponseCommand(QueryConsumerOffsetResponseHeader.class); 5 final QueryConsumerOffsetResponseHeader responseHeader = 6 (QueryConsumerOffsetResponseHeader) response.readCustomHeader(); 7 final QueryConsumerOffsetRequestHeader requestHeader = 8 (QueryConsumerOffsetRequestHeader) request 9 .decodeCommandCustomHeader(QueryConsumerOffsetRequestHeader.class); 10 11 long offset = 12 this.brokerController.getConsumerOffsetManager().queryOffset( 13 requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId()); // @1 14 15 if (offset >= 0) { // @2 16 responseHeader.setOffset(offset); 17 response.setCode(ResponseCode.SUCCESS); 18 response.setRemark(null); 19 } else { // @3 20 long minOffset = 21 this.brokerController.getMessageStore().getMinOffsetInQueue(requestHeader.getTopic(), 22 requestHeader.getQueueId()); // @4 23 if (minOffset <= 0 24 && !this.brokerController.getMessageStore().checkInDiskByConsumeOffset( // @5 25 requestHeader.getTopic(), requestHeader.getQueueId(), 0)) { 26 responseHeader.setOffset(0L); 27 response.setCode(ResponseCode.SUCCESS); 28 response.setRemark(null); 29 } else { // @6 30 response.setCode(ResponseCode.QUERY_NOT_FOUND); 31 response.setRemark("Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first"); 32 } 33 } 34 return response; 35}
代码@1：从消费消息进度文件中查询消息消费进度。
代码@2：如果消息消费进度文件中存储该队列的消息进度，其返回的offset必然会大于等于0，则直接返回该偏移量该客户端，客户端从该偏移量开始消费。
代码@3：如果未从消息消费进度文件中查询到其进度，offset为-1。则首先获取该主题、消息队列当前在Broker服务器中的最小偏移量(@4)。如果小于等于0(返回0则表示该队列的文件还未曾删除过)并且其最小偏移量对应的消息存储在内存中而不是存在磁盘中，则返回偏移量0，这就意味着ConsumeFromWhere中定义的三种枚举类型都不会生效，直接从0开始消费，到这里就能解开其谜团了(@5)。
代码@6：如果偏移量小于等于0，但其消息已经存储在磁盘中，此时返回未找到，最终RebalancePushImpl#computePullFromWhere中得到的偏移量为-1。
看到这里，大家应该能回答文章开头处提到的问题了吧？
看到这里，大家应该明白了，为什么设置的CONSUME_FROM_LAST_OFFSET，但消费组是从消息队列的开始处消费了吧，原因就是消息消费进度文件中并没有找到其消息消费进度，并且该队列在Broker端的最小偏移量为0，说的更直白点，consumequeue/topicName/queueNum的第一个消息消费队列文件为00000000000000000000,并且消息其对应的消息缓存在Broker端的内存中(pageCache)，其返回给消费端的偏移量为0，故会从0开始消费，而不是从队列的最大偏移量处开始消费。
为了知识体系的完备性，我们顺便来看一下其他两种策略的计算逻辑。
CONSUME_FROM_FIRST_OFFSET
1case CONSUME_FROM_FIRST_OFFSET: { 2 long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE); // @1 3 if (lastOffset >= 0) { // @2 4 result = lastOffset; 5 } else if (-1 == lastOffset) { // @3 6 result = 0L; 7 } else { 8 result = -1; // @4 9 } 10 break; 11}
从队列的开始偏移量开始消费，其计算逻辑如下：
代码@1：首先通过偏移量存储器查询消费队列的消费进度。
代码@2：如果大于等于0，则从当前该偏移量开始消费。
代码@3：如果远程返回-1，表示并没有存储该队列的消息消费进度，从0开始。
代码@4：否则从-1开始消费。
CONSUME_FROM_TIMESTAMP
从指定时戳后的消息开始消费。
1case CONSUME_FROM_TIMESTAMP: { 2 ong lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE); // @1 3 if (lastOffset >= 0) { // @2 4 result = lastOffset; 5 } else if (-1 == lastOffset) { // @3 6 if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { 7 try { 8 result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq); 9 } catch (MQClientException e) { 10 result = -1; 11 } 12 } else { 13 try { 14 long timestamp = UtilAll.parseDate(this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp(), 15 UtilAll.YYYYMMDDHHMMSS).getTime(); 16 result = this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp); 17 } catch (MQClientException e) { 18 result = -1; 19 } 20 } 21 } else { 22 result = -1; 23 } 24 break; 25}
其基本套路与CONSUME_FROM_LAST_OFFSET一样：
代码@1：首先通过偏移量存储器查询消费队列的消费进度。
代码@2：如果大于等于0，则从当前该偏移量开始消费。
代码@3：如果远程返回-1，表示并没有存储该队列的消息消费进度，如果是重试主题，则从当前队列的最大偏移量开始消费，如果是普通主题，则根据时间戳去Broker端查询，根据查询到的偏移量开始消费。
原理就介绍到这里，下面根据上述理论对其进行验证。
猜想与验证
根据上述理论分析我们得知设置CONSUME_FROM_LAST_OFFSET但并不是从消息队列的最大偏移量开始消费的“罪魁祸首”是因为消息消费队列的最小偏移量为0，如果不为0，则就会符合预期，我们来验证一下这个猜想。
首先我们删除commitlog目录下的文件，如图所示：
其消费队列截图如下：
消费端的验证代码如下：
1public static void main(String[] args) throws InterruptedException, MQClientException { 2 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_02"); 3 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); 4 consumer.subscribe("TopicTest", "*"); 5 consumer.setNamesrvAddr("127.0.0.1:9876"); 6 consumer.registerMessageListener(new MessageListenerConcurrently() { 7 @Override 8 public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 9 ConsumeConcurrentlyContext context) { 10 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs); 11 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 12 } 13 }); 14 consumer.start(); 15 System.out.printf("Consumer Started.%n"); 16}
运行结果如下：
并没有消息存在的消息，符合预期。
解决方案
如果在生产环境下，一个新的消费组订阅一个已经存在比较久的topic，设置CONSUME_FROM_MAX_OFFSET是符合预期的，即该主题的consumequeue/{queueNum}/fileName，fileName通常不会是00000000000000000000，如是是上面文件名，想要实现从队列的最后开始消费，该如何做呢？那就走自动创建消费组的路子，执行如下命令：
1./mqadmin updateSubGroup -n 127.0.0.1:9876 -c DefaultCluster -g my_consumer_05 2 3//克隆一个订阅了该topic的消费组消费进度 4./mqadmin cloneGroupOffset -n 127.0.0.1:9876 -s my_consumer_01 -d my_consumer_05 -t TopicTest 5 6//重置消费进度到当前队列的最大值 7./mqadmin resetOffsetByTime -n 127.0.0.1:9876 -g my_consumer_05 -t TopicTest -s -1
按照上上述命令后，即可实现其目的。
