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

,大家好，我是君哥。,对消息队列来说，偏移量是一个非常重要的概念，如果偏移量保存失败，可能会造成消息丢失、消息重复消费等问题。今天来聊一聊 RocketMQ 是怎么保存消息偏移量的。,RocketMQ 客户端启动的时候，会启动重平衡线程 RebalanceService，在这里创建拉取消息的请求。下面 UML 类图展示了客户端启动重平衡线程的调用关系：,,业务入口是 MQClientInstance 中的 start 方法，start 方法中启动了 RebalanceService 线程，线程中的 run 方法又返回来调用了 MQClientInstance 中的 doRebalance 方法，最终调用到了 RebalanceImpl 中的 doRebalance 方法。,RebalanceService 的 run 方法是一个死循环，不停地进行重平衡操作，代码如下:,那重平衡操作具体是做什么呢？再看下面的代码：,可以看到，无论是集群模式还是广播模式，updateProcessQueueTableInRebalance 方法最终都被调用了。这个方法封装了拉取消息的请求 PullRequest。这些请求被 put 到 PullMessageService 的 pullRequestQueue，然后 PullMessageService 使用死循环不停地从 pullRequestQueue 中 take 请求 发送到 Broker。这个处理的 UML 类图如下：,,偏移量这个参数封装在消息拉取请求 PullRequest 中，看一下封装拉取请求的代码：,从上面代码可以看出，程序是通过 computePullFromWhereWithException 这个方法获取消息偏移量。下面着重看一下这个方法。,首先，看一下获取偏移量的方法：,CONSUME_FROM_LAST_OFFSET 这个分支是默认的分支，其他情况一般不会用，为了不放大量代码，这里做了省略。,上面的代码看出，偏移量是从 OffsetStore 中获取的。OffsetStore 是一个接口，实现类有两个，如下图：,,那获取偏移量的时候，这两个实现类选择哪一个呢？从 DefaultMQPushConsumerImpl 的 start 方法中可以看出：,从上面的代码可以看到，OffsetStore 的两个实现类对应了广播模式和集群模式。从文件名也可以看到，LocalFileOffsetStore 是从本地读取偏移量，而 RemoteBrokerOffsetStore 则从 Broker 端请求偏移量。,OffsetStore 的两个实现类保存偏移量的数据结构是一样的，如下图：,,从前面的分析可以看到，广播模式的偏移量是保存在本地，分析源码可以看到，文件默认保存在：,可以通过参数【rocketmq.client.localOffsetStoreDir】进行配置，这样文件保存的路径就是：,OffsetStore 的 load 方法读取上面文件，如果读取失败或者文件内容是空，就会读取备份文件，路径是上面的文件名后面加 .bak。,load 方法读取这个 json 文件，然后把内容读取到 LocalFileOffsetStore 类的 offsetTable 这个数据结构中：,上面调用的 readLocalOffset 方法代码如下：,在拉取消息时，首先会封装 PullRequest 请求，PullRequest 中的 nextOffset 参数需要从 offsetTable 获取，代码如下：,上面的代码省略了 READ_FROM_MEMORY 这个 case，因为我跟踪了源代码，广播模式并不会走到这个 case。,MQClientInstance 初始化时，会启动定时任务，每隔 5s 执行一次，把偏移量持久化到本地文件，代码如下：,上面代码最终调用了 LocalFileOffsetStore 类的 persistAll 方法，这个方法把内存中 offsetTable 变量中保存的值写入本地 offsets.json 文件。,写文件时分四步：,总结，广播模式下，偏移量保存在消费者本地服务器。这是因为所有的消费者都要消费同一个队列，消费者维护偏移量会更加方便。,前面分析过，集群模式客户端处理偏移量的类是 RemoteBrokerOffsetStore。,集群模式下，偏移量是从 Broker 端获取，所以客户端 RemoteBrokerOffsetStore 中的 load 方法没有内容。,在 Broker 初始化时，会加载本地的偏移量文件，调用关系的 UML 类图如下：,,BrokerController 初始化代码如下：,这里最终调用了 ConsumerOffsetManager 的 decode 方法，代码如下：,从上面可以看到，Broker 上的偏移量最终保存在 offsetTable 这个变量上，数据结构如下图：,,从源码中可以看到，Broker 端偏移量文件如下：,获取 Broker 端偏移量的时候，会向 Broker 发送一个请求，请求码是 QUERY_CONSUMER_OFFSET，通过请求码就可以找到 Broker 处理的逻辑。代码如下：,上面的处理逻辑如下：,在消费端，定时任务每 5s 向 Broker 发送更新消息偏移量的请求，请求码是 UPDATE_CONSUMER_OFFSET。,Broker 收到后，ConsumerOffsetManager 类更新 offsetTable 变量。,Broker 端也会用定时任务每 5s 从 offsetTable 变量刷到本地文件。逻辑跟消费端的保存逻辑一样，就不再介绍了。,广播模式下，偏移量保存在消费者本地。这也是最合理的，因为每个消费者都要消费同一个 MessageQueue，自己维护自己的偏移量更简单。不过这可能也是广播模式下不支持消息重试的原因，因为如果一个消费者消费失败了，这批消息其他消费成功的消费者也需要重试，导致重复消费。,集群模式下，偏移量保存在 Broker 服务器，消费者需要通过请求的方式来获取和维护偏移量。