写点什么

这次 6 张图带你彻底搞懂 RocketMQ 是怎么保存偏移量的?

作者:程序员小毕
  • 2022 年 8 月 15 日
    湖南
  • 本文字数:6073 字

    阅读完需:约 20 分钟

这次6张图带你彻底搞懂RocketMQ是怎么保存偏移量的?

对消息队列来说,偏移量是一个非常重要的概念,如果偏移量保存失败,可能会造成消息丢失、消息重复消费等问题。今天来聊一聊 RocketMQ 是怎么保存消息偏移量的。

1 消息拉取

RocketMQ 客户端启动的时候,会启动重平衡线程 RebalanceService,在这里创建拉取消息的请求。下面 UML 类图展示了客户端启动重平衡线程的调用关系:


业务入口是 MQClientInstance 中的 start 方法,start 方法中启动了 RebalanceService 线程,线程中的 run 方法又返回来调用了 MQClientInstance 中的 doRebalance 方法,最终调用到了 RebalanceImpl 中的 doRebalance 方法。

RebalanceService 的 run 方法是一个死循环,不停地进行重平衡操作,代码如下:

public void run() { log.info(this.getServiceName() + " service started"); while (!this.isStopped()) {  this.waitForRunning(waitInterval);  this.mqClientFactory.doRebalance(); } log.info(this.getServiceName() + " service end");}
复制代码

那重平衡操作具体是做什么呢?再看下面的代码:

private void rebalanceByTopic(final String topic, final boolean isOrder) { switch (messageModel) {  case BROADCASTING: {   Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);   if (mqSet != null) {    boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);   } else {}   break;  }  case CLUSTERING: {   Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);   List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);   if (mqSet != null && cidAll != null) {    List<MessageQueue> mqAll = new ArrayList<MessageQueue>();    mqAll.addAll(mqSet);    List<MessageQueue> allocateResult = null;    //这里根据负载均衡策略进行获取分配给自己的 MessageQueue,逻辑省略    Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();    if (allocateResult != null) {     allocateResultSet.addAll(allocateResult);    }    boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);   }   break;  }  default:   break; }}
复制代码

可以看到,无论是集群模式还是广播模式,updateProcessQueueTableInRebalance 方法最终都被调用了。这个方法封装了拉取消息的请求 PullRequest。这些请求被 put 到 PullMessageService 的 pullRequestQueue,然后 PullMessageService 使用死循环不停地从 pullRequestQueue 中 take 请求 发送到 Broker。这个处理的 UML 类图如下:


偏移量这个参数封装在消息拉取请求 PullRequest 中,看一下封装拉取请求的代码:

private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet, final boolean isOrder) { boolean changed = false;    //省略掉判断 List<PullRequest> pullRequestList = new ArrayList<PullRequest>(); for (MessageQueue mq : mqSet) {  if (!this.processQueueTable.containsKey(mq)) {            //省略部分逻辑   long nextOffset = -1L;   try {    nextOffset = this.computePullFromWhereWithException(mq);   } //省略catch   if (nextOffset >= 0) {    ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);    if (pre != null) {} else {     PullRequest pullRequest = new PullRequest();     pullRequest.setConsumerGroup(consumerGroup);     pullRequest.setNextOffset(nextOffset);     pullRequest.setMessageQueue(mq);     pullRequest.setProcessQueue(pq);     pullRequestList.add(pullRequest);     changed = true;    }   } //省略 else  } } this.dispatchPullRequest(pullRequestList); return changed;}
复制代码

从上面代码可以看出,程序是通过 computePullFromWhereWithException 这个方法获取消息偏移量。下面着重看一下这个方法。

2 偏移量处理

首先,看一下获取偏移量的方法:

public long computePullFromWhereWithException(MessageQueue mq) throws MQClientException { long result = -1; final ConsumeFromWhere consumeFromWhere = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere(); final OffsetStore offsetStore = this.defaultMQPushConsumerImpl.getOffsetStore(); switch (consumeFromWhere) {  case CONSUME_FROM_LAST_OFFSET: {   long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);   if (lastOffset >= 0) {    result = lastOffset;   }   // First start,no offset   else if (-1 == lastOffset) {    if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {     result = 0L;    } else {     try {      result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);     } catch (MQClientException e) {     }    }   } else {    result = -1;   }   break;  }  default:   break; } return result;}
复制代码

CONSUME_FROM_LAST_OFFSET 这个分支是默认的分支,其他情况一般不会用,为了不放大量代码,这里做了省略。

上面的代码看出,偏移量是从 OffsetStore 中获取的。OffsetStore 是一个接口,实现类有两个,如下图:


那获取偏移量的时候,这两个实现类选择哪一个呢?从 DefaultMQPushConsumerImpl 的 start 方法中可以看出:

public synchronized void start() throws MQClientException { switch (this.serviceState) {  case CREATE_JUST:   if (this.defaultMQPushConsumer.getOffsetStore() != null) {    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();   } else {    switch (this.defaultMQPushConsumer.getMessageModel()) {     case BROADCASTING:      this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());      break;     case CLUSTERING:      this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());      break;     default:      break;    }    this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);   }   this.offsetStore.load();   break;  default:   break; }}
复制代码

从上面的代码可以看到,OffsetStore 的两个实现类对应了广播模式和集群模式。从文件名也可以看到,LocalFileOffsetStore 是从本地读取偏移量,而 RemoteBrokerOffsetStore 则从 Broker 端请求偏移量。

OffsetStore 的两个实现类保存偏移量的数据结构是一样的,如下图:


3 广播模式

从前面的分析可以看到,广播模式的偏移量是保存在本地,分析源码可以看到,文件默认保存在:

/home/${user}/.rocketmq_offsets/${clientId}/${groupName}/offsets.json
复制代码

可以通过参数【rocketmq.client.localOffsetStoreDir】进行配置,这样文件保存的路径就是:

/${rocketmq.client.localOffsetStoreDir}/.rocketmq_offsets/${clientId}/${groupName}/offsets.json
复制代码

OffsetStore 的 load 方法读取上面文件,如果读取失败或者文件内容是空,就会读取备份文件,路径是上面的文件名后面加 .bak。


相关视频解析:RocketMQ源码解析

3.1 加载偏移量

load 方法读取这个 json 文件,然后把内容读取到 LocalFileOffsetStore 类的 offsetTable 这个数据结构中:

//LocalFileOffsetStore类public void load() throws MQClientException { OffsetSerializeWrapper offsetSerializeWrapper = this.readLocalOffset(); if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) {  offsetTable.putAll(offsetSerializeWrapper.getOffsetTable()); }}
复制代码

上面调用的 readLocalOffset 方法代码如下:

private OffsetSerializeWrapper readLocalOffset() throws MQClientException { String content = null; try {  content = MixAll.file2String(this.storePath); } catch (IOException e) {  log.warn("Load local offset store file exception", e); } if (null == content || content.length() == 0) {     //读取失败或者文件内容是空,则从 .bak 文件获取  return this.readLocalOffsetBak(); } else {  OffsetSerializeWrapper offsetSerializeWrapper = null;  try {   offsetSerializeWrapper =    OffsetSerializeWrapper.fromJson(content, OffsetSerializeWrapper.class);  } catch (Exception e) {   log.warn("readLocalOffset Exception, and try to correct", e);   return this.readLocalOffsetBak();  }  return offsetSerializeWrapper; }}
复制代码

3.2 读取偏移量

在拉取消息时,首先会封装 PullRequest 请求,PullRequest 中的 nextOffset 参数需要从 offsetTable 获取,代码如下:

public long readOffset(final MessageQueue mq, final ReadOffsetType type) { if (mq != null) {  switch (type) {   //省略其他 case   case READ_FROM_STORE: {    OffsetSerializeWrapper offsetSerializeWrapper;    try {     offsetSerializeWrapper = this.readLocalOffset();    } catch (MQClientException e) {     return -1;    }    if (offsetSerializeWrapper != null && offsetSerializeWrapper.getOffsetTable() != null) {     AtomicLong offset = offsetSerializeWrapper.getOffsetTable().get(mq);     if (offset != null) {         //新读取到的偏移量添加到 offsetTable 中      this.updateOffset(mq, offset.get(), false);      return offset.get();     }    }   }   default:    break;  } } return -1;}
复制代码

上面的代码省略了 READ_FROM_MEMORY 这个 case,因为我跟踪了源代码,广播模式并不会走到这个 case。

3.3 更新偏移量

MQClientInstance 初始化时,会启动定时任务,每隔 5s 执行一次,把偏移量持久化到本地文件,代码如下:

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() {  try {   MQClientInstance.this.persistAllConsumerOffset();  } catch (Exception e) {   log.error("ScheduledTask persistAllConsumerOffset exception", e);  } }}, 1000 * 10, this.clientConfig.getPersistConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
复制代码

上面代码最终调用了 LocalFileOffsetStore 类的 persistAll 方法,这个方法把内存中 offsetTable 变量中保存的值写入本地 offsets.json 文件。

写文件时分四步:

  1. 首先把内容写入到 offsets.json.tmp 文件;

  2. offsets.json 内容备份到 offsets.json.bak;

  3. 删除 offsets.json 文件;

  4. 把 offsets.json.tmp 改名为 offsets.json。

总结,广播模式下,偏移量保存在消费者本地服务器。这是因为所有的消费者都要消费同一个队列,消费者维护偏移量会更加方便。

4 集群模式

前面分析过,集群模式客户端处理偏移量的类是 RemoteBrokerOffsetStore。

4.1 加载偏移量

集群模式下,偏移量是从 Broker 端获取,所以客户端 RemoteBrokerOffsetStore 中的 load 方法没有内容。

在 Broker 初始化时,会加载本地的偏移量文件,调用关系的 UML 类图如下:


BrokerController 初始化代码如下:

//BrokerController.javapublic boolean initialize() throws CloneNotSupportedException {    //省略其他代码 result = result && this.consumerOffsetManager.load();}
复制代码

这里最终调用了 ConsumerOffsetManager 的 decode 方法,代码如下:

public void decode(String jsonString) { if (jsonString != null) {  ConsumerOffsetManager obj = RemotingSerializable.fromJson(jsonString, ConsumerOffsetManager.class);  if (obj != null) {   this.offsetTable = obj.offsetTable;  } }}
复制代码

从上面可以看到,Broker 上的偏移量最终保存在 offsetTable 这个变量上,数据结构如下图:


从源码中可以看到,Broker 端偏移量文件如下:

/home/${user}/store/config/consumerOffset.json
复制代码

4.2 获取偏移量

获取 Broker 端偏移量的时候,会向 Broker 发送一个请求,请求码是 QUERY_CONSUMER_OFFSET,通过请求码就可以找到 Broker 处理的逻辑。代码如下:

private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { //省略变量定义 //从 offsetTable 变量中查找 long offset =  this.brokerController.getConsumerOffsetManager().queryOffset(   requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId()); if (offset >= 0) {  responseHeader.setOffset(offset);  response.setCode(ResponseCode.SUCCESS);  response.setRemark(null); } else {  long minOffset =   this.brokerController.getMessageStore().getMinOffsetInQueue(requestHeader.getTopic(),    requestHeader.getQueueId());  if (minOffset <= 0   && !this.brokerController.getMessageStore().checkInDiskByConsumeOffset(   requestHeader.getTopic(), requestHeader.getQueueId(), 0)) {   responseHeader.setOffset(0L);   response.setCode(ResponseCode.SUCCESS);   response.setRemark(null);  } else {   response.setCode(ResponseCode.QUERY_NOT_FOUND);   response.setRemark("Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first");  } } return response;}
复制代码

上面的处理逻辑如下:

  1. 首先从 offsetTable 变量中查找,如果找到了就直接返回给消费者;

  2. 如果没有找到,则从 Broker 上查找这个 MessageQueue 的最小偏移量,如果偏移量小于等于 0 并且没有被交换到磁盘上(保存在内存里),则返回偏移量是 0;

  3. 否则返回查找失败。

4.3 更新偏移量

在消费端,定时任务每 5s 向 Broker 发送更新消息偏移量的请求,请求码是 UPDATE_CONSUMER_OFFSET。

Broker 收到后,ConsumerOffsetManager 类更新 offsetTable 变量。

Broker 端也会用定时任务每 5s 从 offsetTable 变量刷到本地文件。逻辑跟消费端的保存逻辑一样,就不再介绍了。

5 总结

广播模式下,偏移量保存在消费者本地。这也是最合理的,因为每个消费者都要消费同一个 MessageQueue,自己维护自己的偏移量更简单。不过这可能也是广播模式下不支持消息重试的原因,因为如果一个消费者消费失败了,这批消息其他消费成功的消费者也需要重试,导致重复消费。

集群模式下,偏移量保存在 Broker 服务器,消费者需要通过请求的方式来获取和维护偏移量。

原文:https://mp.weixin.qq.com/s/4t-40pHeevjoKhw3UOUpVA

如果感觉本文对你有帮助,点赞关注支持一下,想要了解更多 Java 后端,大数据,算法领域最新资讯可以关注我公众号【架构师老毕】私信 666 还可获取更多 Java 后端,大数据,算法 PDF+大厂最新面试题整理+视频精讲

用户头像

领取资料添加小助理vx:bjmsb2020 2020.12.19 加入

Java领域;架构知识;面试心得;互联网行业最新资讯

评论

发布
暂无评论
这次6张图带你彻底搞懂RocketMQ是怎么保存偏移量的?_Java_程序员小毕_InfoQ写作社区