六张图告诉你 RocketMQ 是怎么保存偏移量的

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

大家好，我是君哥。

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

一、消息拉取

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");
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    •

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

复制 
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;
 }
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    • 

								
  • 
									11.
								
    • 

								
  • 
									12.
								
    • 

								
  • 
									13.
								
    • 

								
  • 
									14.
								
    • 

								
  • 
									15.
								
    • 

								
  • 
									16.
								
    • 

								
  • 
									17.
								
    • 

								
  • 
									18.
								
    • 

								
  • 
									19.
								
    • 

								
  • 
									20.
								
    • 

								
  • 
									21.
								
    • 

								
  • 
									22.
								
    • 

								
  • 
									23.
								
    • 

								
  • 
									24.
								
    • 

								
  • 
									25.
								
    • 

								
  • 
									26.
								
    • 

								
  • 
									27.
								
    • 

								
  • 
									28.
								
    • 

								
  • 
									29.
								
    • 

								
  • 
									30.
								
    •

可以看到，无论是集群模式还是广播模式，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;
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    • 

								
  • 
									11.
								
    • 

								
  • 
									12.
								
    • 

								
  • 
									13.
								
    • 

								
  • 
									14.
								
    • 

								
  • 
									15.
								
    • 

								
  • 
									16.
								
    • 

								
  • 
									17.
								
    • 

								
  • 
									18.
								
    • 

								
  • 
									19.
								
    • 

								
  • 
									20.
								
    • 

								
  • 
									21.
								
    • 

								
  • 
									22.
								
    • 

								
  • 
									23.
								
    • 

								
  • 
									24.
								
    • 

								
  • 
									25.
								
    • 

								
  • 
									26.
								
    • 

								
  • 
									27.
								
    • 

								
  • 
									28.
								
    • 

								
  • 
									29.
								
    •

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

二、偏移量处理

首先，看一下获取偏移量的方法：

复制 
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;
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    • 

								
  • 
									11.
								
    • 

								
  • 
									12.
								
    • 

								
  • 
									13.
								
    • 

								
  • 
									14.
								
    • 

								
  • 
									15.
								
    • 

								
  • 
									16.
								
    • 

								
  • 
									17.
								
    • 

								
  • 
									18.
								
    • 

								
  • 
									19.
								
    • 

								
  • 
									20.
								
    • 

								
  • 
									21.
								
    • 

								
  • 
									22.
								
    • 

								
  • 
									23.
								
    • 

								
  • 
									24.
								
    • 

								
  • 
									25.
								
    • 

								
  • 
									26.
								
    • 

								
  • 
									27.
								
    • 

								
  • 
									28.
								
    • 

								
  • 
									29.
								
    • 

								
  • 
									30.
								
    •

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;
 }
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    • 

								
  • 
									11.
								
    • 

								
  • 
									12.
								
    • 

								
  • 
									13.
								
    • 

								
  • 
									14.
								
    • 

								
  • 
									15.
								
    • 

								
  • 
									16.
								
    • 

								
  • 
									17.
								
    • 

								
  • 
									18.
								
    • 

								
  • 
									19.
								
    • 

								
  • 
									20.
								
    • 

								
  • 
									21.
								
    • 

								
  • 
									22.
								
    • 

								
  • 
									23.
								
    • 

								
  • 
									24.
								
    •

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

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

图片图片

三、广播模式

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

复制 
/home/${user}/.rocketmq_offsets/${clientId}/${groupName}/offsets.json
							
    
								
								
  • 
									1.
								
    •

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

复制 
/${rocketmq.client.localOffsetStoreDir}/.rocketmq_offsets/${clientId}/${groupName}/offsets.json
							
    
								
								
  • 
									1.
								
    •

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

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());
 }
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    •

上面调用的 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;
 }
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    • 

								
  • 
									11.
								
    • 

								
  • 
									12.
								
    • 

								
  • 
									13.
								
    • 

								
  • 
									14.
								
    • 

								
  • 
									15.
								
    • 

								
  • 
									16.
								
    • 

								
  • 
									17.
								
    • 

								
  • 
									18.
								
    • 

								
  • 
									19.
								
    • 

								
  • 
									20.
								
    • 

								
  • 
									21.
								
    • 

								
  • 
									22.
								
    • 

								
  • 
									23.
								
    •

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;
}
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    • 

								
  • 
									11.
								
    • 

								
  • 
									12.
								
    • 

								
  • 
									13.
								
    • 

								
  • 
									14.
								
    • 

								
  • 
									15.
								
    • 

								
  • 
									16.
								
    • 

								
  • 
									17.
								
    • 

								
  • 
									18.
								
    • 

								
  • 
									19.
								
    • 

								
  • 
									20.
								
    • 

								
  • 
									21.
								
    • 

								
  • 
									22.
								
    • 

								
  • 
									23.
								
    • 

								
  • 
									24.
								
    • 

								
  • 
									25.
								
    • 

								
  • 
									26.
								
    •

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

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);
							
    
								
								
  • 
									1.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    •

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

写文件时分四步：

  1. 首先把内容写入到 offsets.json.tmp 文件。
  2. offsets.json 内容备份到 offsets.json.bak。
  3. 删除 offsets.json 文件。
  4. 把 offsets.json.tmp 改名为 offsets.json。

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

四、集群模式

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

1、加载偏移量

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

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

图片

BrokerController 初始化代码如下：

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

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    •

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

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

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    •

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

图片

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

复制 
/home/${user}/store/config/consumerOffset.json
							
    
								
								
  • 
									1.
								
    •

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.
								
    • 

								
  • 
									2.
								
    • 

								
  • 
									3.
								
    • 

								
  • 
									4.
								
    • 

								
  • 
									5.
								
    • 

								
  • 
									6.
								
    • 

								
  • 
									7.
								
    • 

								
  • 
									8.
								
    • 

								
  • 
									9.
								
    • 

								
  • 
									10.
								
    • 

								
  • 
									11.
								
    • 

								
  • 
									12.
								
    • 

								
  • 
									13.
								
    • 

								
  • 
									14.
								
    • 

								
  • 
									15.
								
    • 

								
  • 
									16.
								
    • 

								
  • 
									17.
								
    • 

								
  • 
									18.
								
    • 

								
  • 
									19.
								
    • 

								
  • 
									20.
								
    • 

								
  • 
									21.
								
    • 

								
  • 
									22.
								
    • 

								
  • 
									23.
								
    • 

								
  • 
									24.
								
    • 

								
  • 
									25.
								
    • 

								
  • 
									26.
								
    • 

								
  • 
									27.
								
    • 

								
  • 
									28.
								
    •

上面的处理逻辑如下：

  1. 首先从 offsetTable 变量中查找，如果找到了就直接返回给消费者。
  2. 如果没有找到，则从 Broker 上查找这个 MessageQueue 的最小偏移量，如果偏移量小于等于 0 并且没有被交换到磁盘上（保存在内存里），则返回偏移量是 0。
  3. 否则返回查找失败。

3、更新偏移量

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

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

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

五、总结

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

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

责任编辑：姜华 来源： 君哥聊技术

标签：

网站建设橙云橙邮我要创业网橙米网网络知识
声明：本站发布的内容以原创、转载、分享网络内容为主，如有侵权，请联系电话：021-51697771-8029，邮箱：mj@cndns.com ，我们将会在第一时间删除。文章观点不代表本站立场，如需处理请联系我们。
>　返回列表　<

热门TAG

SEO设计域名云计算邮局电商美橙营销推广网站建设互联网

热门视频

互联网营销时代