RocketMQ一个新的消费组初次启动时从何处开始消费呢？

时间  2019-11-06
标签 rocketmq 一个 新的 消费 初次 启动 何处 开始 繁體版
原文   原文链接

一、抛出问题

一个新的消费组订阅一个已存在的Topic主题时，消费组是从该Topic的哪条消息开始消费呢？java

首先翻阅DefaultMQPushConsumer的API时，setConsumeFromWhere(ConsumeFromWhere consumeFromWhere)API映入眼帘，从字面意思来看是设置消费者从哪里开始消费，正是解开该问题的”钥匙“。ConsumeFromWhere枚举类图以下：缓存

在这里插入图片描述

  • CONSUME_FROM_MAX_OFFSET 从消费队列最大的偏移量开始消费。
  • CONSUME_FROM_FIRST_OFFSET 从消费队列最小偏移量开始消费。
  • CONSUME_FROM_TIMESTAMP 从指定的时间戳开始消费，默认为消费者启动以前的30分钟处开始消费。能够经过DefaultMQPushConsumer#setConsumeTimestamp。

是否是点小激动，还不快试试。服务器

需求：新的消费组启动时，从队列最后开始消费，即只消费启动后发送到消息服务器后的最新消息。ide

1.1 环境准备

本示例所用到的Topic路由信息以下： 在这里插入图片描述测试

Broker的配置以下(broker.conf)this

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH

storePathRootDir=E:/SH2019/tmp/rocketmq_home/rocketmq4.5_simple/store
storePathCommitLog=E:/SH2019/tmp/rocketmq_home/rocketmq4.5_simple/store/commitlog
namesrvAddr=127.0.0.1:9876
autoCreateTopicEnable=false
mapedFileSizeCommitLog=10240
mapedFileSizeConsumeQueue=2000

其中重点修改了以下两个参数：.net

  • mapedFileSizeCommitLog 单个commitlog文件的大小，这里使用10M，方便测试用。
  • mapedFileSizeConsumeQueue 单个consumequeue队列长度，这里使用1000，表示一个consumequeue文件中包含1000个条目。

1.2 消息发送者代码

public static void main(String[] args) throws MQClientException, InterruptedException {
    DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    producer.setNamesrvAddr("127.0.0.1:9876");
    producer.start();
    for (int i = 0; i < 300; i++) {
        try {
            Message msg = new Message("TopicTest" ,"TagA" , ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        } catch (Exception e) {
            e.printStackTrace();
            Thread.sleep(1000);
        }
    }
    producer.shutdown();
}

经过上述，往TopicTest发送300条消息，发送完毕后，RocketMQ Broker存储结构以下： 在这里插入图片描述code

1.3 消费端验证代码

public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_01");
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    consumer.subscribe("TopicTest", "*");
    consumer.setNamesrvAddr("127.0.0.1:9876");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        [@Override](https://my.oschina.net/u/1162528)
        public ConsumeConcurrentlyStatus consumeMessage(List<messageext> msgs,
            ConsumeConcurrentlyContext context) {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.out.printf("Consumer Started.%n");
}

执行上述代码后，按照指望，应该是不会消费任何消息，只有等生产者再发送消息后，才会对消息进行消费，事实是这样吗？执行效果如图所示： 在这里插入图片描述中间件

使人意外的是，居然从队列的最小偏移量开始消费了，这就“尴尬”了。难不成是RocketMQ的Bug。带着这个疑问，从源码的角度尝试来解读该问题，并指导咱们实践。blog

二、探究CONSUME_FROM_MAX_OFFSET实现原理

对于一个新的消费组，不管是集群模式仍是广播模式都不会存储该消费组的消费进度，能够理解为-1,此时就须要根据DefaultMQPushConsumer#consumeFromWhere属性来决定其从何处开始消费，首先咱们须要找到其对应的处理入口。咱们知道，消息消费者从Broker服务器拉取消息时，须要进行消费队列的负载，即RebalanceImpl。

> 舒适提示：本文不会详细介绍RocketMQ消息队列负载、消息拉取、消息消费逻辑，只会展现出通往该问题的简短流程，如想详细了解消息消费具体细节，建议购买笔者出版的《RocketMQ技术内幕》书籍。

RebalancePushImpl#computePullFromWhere

public long computePullFromWhere(MessageQueue mq) {
        long result = -1;                                                                                                                                                                                                                  // [@1](https://my.oschina.net/u/1198)
        final ConsumeFromWhere consumeFromWhere = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere();    
        final OffsetStore offsetStore = this.defaultMQPushConsumerImpl.getOffsetStore();
        switch (consumeFromWhere) {
            case CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST:
            case CONSUME_FROM_MIN_OFFSET:
            case CONSUME_FROM_MAX_OFFSET:
            case CONSUME_FROM_LAST_OFFSET: {                                                                                                                                                                // @2
               // 省略部分代码
                break;
            }
            case CONSUME_FROM_FIRST_OFFSET: {                                                                                                                                                              // [@3](https://my.oschina.net/u/2648711)
                // 省略部分代码
                break;
            }
            case CONSUME_FROM_TIMESTAMP: {                                                                                                                                                                  //@4
                // 省略部分代码
                break;
            }
            default:
                break;
        }
        return result;                                                                                                                                                                                                                  // @5
    }

代码@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：返回最后计算的偏移量，从该偏移量出开始消费。

2.1 CONSUME_FROM_LAST_OFFSET计算逻辑

case CONSUME_FROM_LAST_OFFSET: {
    long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);   // @1
    if (lastOffset &gt;= 0) {                                                                                                             // @2
        result = lastOffset;
    }
    // First start,no offset
    else if (-1 == lastOffset) {                                                                                                  // @3
        if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {               
            result = 0L;
        } else {
            try {
                result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);                     
            } catch (MQClientException e) {                                                                              // @4
                result = -1;
            }
        }
    } else {
        result = -1;    
    }
	break;
}

代码@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。

ConsumerManageProcessor#queryConsumerOffset
private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response =
        RemotingCommand.createResponseCommand(QueryConsumerOffsetResponseHeader.class);
    final QueryConsumerOffsetResponseHeader responseHeader =
        (QueryConsumerOffsetResponseHeader) response.readCustomHeader();
    final QueryConsumerOffsetRequestHeader requestHeader =
        (QueryConsumerOffsetRequestHeader) request
            .decodeCommandCustomHeader(QueryConsumerOffsetRequestHeader.class);

    long offset =
        this.brokerController.getConsumerOffsetManager().queryOffset(
            requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());    // @1

    if (offset &gt;= 0) {                                                                                                                                          // @2
        responseHeader.setOffset(offset);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
    } else {                                                                                                                                                       // @3
        long minOffset =
            this.brokerController.getMessageStore().getMinOffsetInQueue(requestHeader.getTopic(),
                requestHeader.getQueueId());                                                                                                     // @4
        if (minOffset &lt;= 0
            &amp;&amp; !this.brokerController.getMessageStore().checkInDiskByConsumeOffset(                                // @5
            requestHeader.getTopic(), requestHeader.getQueueId(), 0)) {
            responseHeader.setOffset(0L);
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);
        } else {                                                                                                                                                 // @6
            response.setCode(ResponseCode.QUERY_NOT_FOUND);
            response.setRemark("Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first");
        }
    }
    return response;
}

代码@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开始消费，而不是从队列的最大偏移量处开始消费。

为了知识体系的完备性，咱们顺便来看一下其余两种策略的计算逻辑。

2.2 CONSUME_FROM_FIRST_OFFSET

case CONSUME_FROM_FIRST_OFFSET: {
    long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);   // @1
    if (lastOffset &gt;= 0) {    // @2
        result = lastOffset;
    } else if (-1 == lastOffset) {  // @3
        result = 0L;
    } else {                                  
        result = -1;                    // @4
    }
    break;
}

从队列的开始偏移量开始消费，其计算逻辑以下： 代码@1：首先经过偏移量存储器查询消费队列的消费进度。

代码@2：若是大于等于0，则从当前该偏移量开始消费。

代码@3：若是远程返回-1，表示并无存储该队列的消息消费进度，从0开始。

代码@4：不然从-1开始消费。

2.4 CONSUME_FROM_TIMESTAMP

从指定时戳后的消息开始消费。

case CONSUME_FROM_TIMESTAMP: {
    ong lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);   // @1
    if (lastOffset &gt;= 0) {                                                                                                            // @2
        result = lastOffset;
    } else if (-1 == lastOffset) {                                                                                                 // @3
        if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
            try {
                result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
            } catch (MQClientException e) {
                result = -1;
            }
        } else {
            try {
                long timestamp = UtilAll.parseDate(this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp(),
                    UtilAll.YYYYMMDDHHMMSS).getTime();
                result = this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
            } catch (MQClientException e) {
                result = -1;
            }
        }
    } else {
        result = -1;
    }
    break;
}

其基本套路与CONSUME_FROM_LAST_OFFSET同样： 代码@1：首先经过偏移量存储器查询消费队列的消费进度。

代码@2：若是大于等于0，则从当前该偏移量开始消费。

代码@3：若是远程返回-1，表示并无存储该队列的消息消费进度，若是是重试主题，则从当前队列的最大偏移量开始消费，若是是普通主题，则根据时间戳去Broker端查询，根据查询到的偏移量开始消费。

原理就介绍到这里，下面根据上述理论对其进行验证。

三、猜测与验证

根据上述理论分析咱们得知设置CONSUME_FROM_LAST_OFFSET但并非从消息队列的最大偏移量开始消费的“罪魁祸首”是由于消息消费队列的最小偏移量为0，若是不为0，则就会符合预期，咱们来验证一下这个猜测。 首先咱们删除commitlog目录下的文件，如图所示： 在这里插入图片描述

其消费队列截图以下： 在这里插入图片描述

消费端的验证代码以下：

public static void main(String[] args) throws InterruptedException, MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_consumer_02");
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    consumer.subscribe("TopicTest", "*");
    consumer.setNamesrvAddr("127.0.0.1:9876");
    consumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<messageext> msgs,
            ConsumeConcurrentlyContext context) {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
    consumer.start();
    System.out.printf("Consumer Started.%n");
}

运行结果以下： 在这里插入图片描述

并无消息存在的消息，符合预期。

四、解决方案

若是在生产环境下，一个新的消费组订阅一个已经存在比较久的topic，设置CONSUME_FROM_MAX_OFFSET是符合预期的，即该主题的consumequeue/{queueNum}/fileName，fileName一般不会是00000000000000000000，如是是上面文件名，想要实现从队列的最后开始消费，该如何作呢？那就走自动建立消费组的路子，执行以下命令：

./mqadmin updateSubGroup -n 127.0.0.1:9876 -c DefaultCluster -g my_consumer_05

//克隆一个订阅了该topic的消费组消费进度
./mqadmin cloneGroupOffset -n 127.0.0.1:9876 -s my_consumer_01 -d my_consumer_05 -t TopicTest

//重置消费进度到当前队列的最大值
./mqadmin resetOffsetByTime -n 127.0.0.1:9876 -g my_consumer_05 -t TopicTest -s -1

按照上上述命令后，便可实现其目的。

> 做者简介：《RocketMQ技术内幕》做者，RocketMQ 社区布道师，维护公众号：中间件兴趣圈，可扫描以下二维码与做者进行互动。

在这里插入图片描述</messageext></messageext>

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程