对消息队列来说偏移量是一个非常重要的概念如果偏移量保存失败可能会造成消息丢失、消息重复消费等问题。今天来聊一聊 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: { SetMessageQueue mqSet this.topicSubscribeInfoTable.get(topic); if (mqSet ! null) { boolean changed this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder); } else {} break; } case CLUSTERING: { SetMessageQueue mqSet this.topicSubscribeInfoTable.get(topic); ListString cidAll this.mQClientFactory.findConsumerIdList(topic, consumerGroup); if (mqSet ! null cidAll ! null) { ListMessageQueue mqAll new ArrayListMessageQueue(); mqAll.addAll(mqSet); ListMessageQueue allocateResult null; //这里根据负载均衡策略进行获取分配给自己的 MessageQueue逻辑省略 SetMessageQueue allocateResultSet new HashSetMessageQueue(); 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 SetMessageQueue mqSet, final boolean isOrder) { boolean changed false; //省略掉判断 ListPullRequest pullRequestList new ArrayListPullRequest(); 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.jsonOffsetStore 的 load 方法读取上面文件如果读取失败或者文件内容是空就会读取备份文件路径是上面的文件名后面加 .bak。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 文件。写文件时分四步首先把内容写入到 offsets.json.tmp 文件offsets.json 内容备份到 offsets.json.bak删除 offsets.json 文件把 offsets.json.tmp 改名为 offsets.json。总结广播模式下偏移量保存在消费者本地服务器。这是因为所有的消费者都要消费同一个队列消费者维护偏移量会更加方便。4 集群模式前面分析过集群模式客户端处理偏移量的类是 RemoteBrokerOffsetStore。4.1 加载偏移量集群模式下偏移量是从 Broker 端获取所以客户端 RemoteBrokerOffsetStore 中的 load 方法没有内容。在 Broker 初始化时会加载本地的偏移量文件调用关系的 UML 类图如下BrokerController 初始化代码如下//BrokerController.java public 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.json4.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; }上面的处理逻辑如下首先从 offsetTable 变量中查找如果找到了就直接返回给消费者如果没有找到则从 Broker 上查找这个 MessageQueue 的最小偏移量如果偏移量小于等于 0 并且没有被交换到磁盘上保存在内存里则返回偏移量是 0否则返回查找失败。4.3 更新偏移量在消费端定时任务每 5s 向 Broker 发送更新消息偏移量的请求请求码是 UPDATE_CONSUMER_OFFSET。Broker 收到后ConsumerOffsetManager 类更新 offsetTable 变量。Broker 端也会用定时任务每 5s 从 offsetTable 变量刷到本地文件。逻辑跟消费端的保存逻辑一样就不再介绍了。5 总结广播模式下偏移量保存在消费者本地。这也是最合理的因为每个消费者都要消费同一个 MessageQueue自己维护自己的偏移量更简单。不过这可能也是广播模式下不支持消息重试的原因因为如果一个消费者消费失败了这批消息其他消费成功的消费者也需要重试导致重复消费。集群模式下偏移量保存在 Broker 服务器消费者需要通过请求的方式来获取和维护偏移量。