RocketMQ Consumer如何获取并维护消费进度？
作者：互联网
背景
Cosumer消息消费流程比较复杂，比较重要的有下面几个模块：维护消费进度，查找消息，消息过滤，负载均衡，消息处理，回发确认等。限于篇幅，这篇文章主要介绍Consumer是如何获取并维护消费进度。由于以上几个步骤都是紧密相连的，可能会出现互相穿插的情况。
消费进度文件
我们之前的文章RocketMQ 如何构建ConsumerQueue的？中讲过，通过一个服务线程异步的从CommitLog中获取已经写入的消息，然后将消息位置，大小，tagsCode等关键信息保存至选好的ConsumerQueue中。ConsumerQueue是一个索引文件，保存了某个topic下面的消息在CommitLog中的位置等信息。我们消费消息，先从这个ConsumerQueue中读取消息位置，然后再去CommitLog中取消息，这是典型的空间换时间的做法。与此同时，我们需要记录下我们读取到哪里了，下次读取的时候就从上次结束的地方继续往前读，所以我们还需要一个保存消费进度的文件。在RocketMQ中这个文件就是ConsumserOffset.json，在store/config目录下，如下所示：
consumerOffset.json中的内容类似于这样：
其中test_url@sub_localtest是主键，规则是topic@consumerGroup，内容就是每个ConsumerQueue的消费进度。例如0号Queue下个应该消费2599，1号Queue下个应该消费2602，6号Queue下个应该消费102。这个进度是按1累加的，对应于ConsumerQueue的固定的20个字节。
现在我们做个实验，首先通过Producer发送一条消息(代码很简单，就不贴了)，发送结果如下所示：
SendResult [sendStatus=SEND_OK, msgId=C0A84D05091058644D4664E1427C0000, offsetMsgId=C0A84D0000002A9F00000000001ED9BA, messageQueue=MessageQueue [topic=test_url, brokerName=broker-a, queueId=6], queueOffset=102]
从结果中可以看到，消息保存在了broker-a的MesssageQueue-6中的102号位置。然后我们启动Consumer，消费这条消息：
2020-01-20 14:08:04.230 INFO [MessageThread_3] c.c.r.example.simplest.Consumer - 收到消息:topic=test_url,msgId=C0A84D05091058644D4664E1427C0000
消息成功被消费了，然后我们再去看看ConsumerOffset.json文件内容：
看到没？上次截图中，6号Queue的消费进度是102，这次变成了103，已经成功消费了一条消息。
现在我们就要开始深究了：在Consumer消费消息时，这个消费进度是如何获取并维护的呢？
从何处开始消费？
Consumer在通过调用
consumer.start();
启动的时候，会加载消费进度，如下所示：
//DefaultMQPushConsumerImpl.start()方法
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()); // 集群模式下生成一个RemoteBrokerOffsetSotre，消费进度就保存在broker端
break;
default:
break;
}
this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
}
this.offsetStore.load(); // 加载本地进度
上面提到了offsetStore对象，这是一个接口，如下所示：
public interface OffsetStore {
void load() throws MQClientException;
void updateOffset(final MessageQueue mq, final long offset, final boolean increaseOnly);
long readOffset(final MessageQueue mq, final ReadOffsetType type);
void persistAll(final Set<MessageQueue> mqs);
void persist(final MessageQueue mq);
void removeOffset(MessageQueue mq);
Map<MessageQueue, Long> cloneOffsetTable(String topic);
void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException;
}
OffsetStore提供了操作消费进度的方法，例如：加载消费进度，读取消费进度，更新消费进度等等。在集群消费模式下，消费进度并没有持久化在Consumer端，而是保存在了远程Broker端，例如上面使用的是RemoteBrokerOffsetStore类：
public class RemoteBrokerOffsetStore implements OffsetStore {
private final static InternalLogger log = ClientLogger.getLog();
private final MQClientInstance mQClientFactory; // 客户端实例
private final String groupName; // 集群名称
private ConcurrentMap<MessageQueue, AtomicLong> offsetTable =
new ConcurrentHashMap<MessageQueue, AtomicLong>(); // 每个Queue的消费进度
}
因为消费进度保存在broker端，用于加载本地消费进度的load方法在RemoteBrokerOffsetStore中是空的，不做任何事，真正读取消费进度是通过readOffset方法实现的。
目前为止我们知道了，消费进度是存在broker端的一个consumerOffset.json文件中的，通过readOffset方法读取这个文件就知道了从哪里开始消费。
负载均衡简单介绍
在深入了解读取消费进度的readOffset方法前，需要简单了解何时会调用这个方法。
我们文章开头提到Consumer消费消息时的模块，其中包括了负载均衡这个模块。一个topic有多个消费队列，而同一组group下面也可能有多个Consumer订阅了这个topic，于是这些队列需要按照一定策略分配给同一个组下面的Consumer消费，例如下面的平均分配：
在上图中，TOPIC_A有5个队列，有2个Consumer订阅了，如果按照平均分配的话，那就是Consumer1消费其中3个队列，Consumer2消费其中2个队列。这个就是负载均衡。
一个Consumer分配到了几个消息队列，就会相应的创建几个消息处理队列（ProcessQueue，消费消息时会用到），并且此时会生成一个拉取消息的请求（PullRequest，请求消息时会用到），这个请求不是真正的发往broker端的获取消息的请求，而是保存在一个阻塞队列里面，然后由专门的拉取消息的服务线程读取它并组装获取消息请求，发送给broker端（这么做当然是为了获得异步的好处）。这个请求PullRequest里面就临时保存着下个消费进度，如下所示：
只有这里才会生成一个PullRequest，后续该consumerGroup下该Queue的所有拉取消息都是重复使用该PullRequest，只是更新了其中的nextOffset等参数。PullRequest类如下所示：
public class PullRequest {
private String consumerGroup; // 消费分组
private MessageQueue messageQueue; // 消费队列
private ProcessQueue processQueue; // 消息处理队列
private long nextOffset; // 消费进度
private boolean lockedFirst = false; // 是否锁住
}
计算消费进度
在生成PullRequest时，就会计算从何处开始消费（computPullFromWhere）。RocketMQ默认的消费起始点是CONSUME_FROM_LAST_OFFSET（从上个offset开始消费），因此在computPullFromWhere方法中，走到的是这个case:
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) { // 初次启动，broker没有偏移，readOffset返回-1
if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
result = 0L; // 消息重试
} else {
try {
result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);// 获取该消息队列消费的最大偏移
} catch (MQClientException e) {
result = -1;
}
}
} else {
result = -1; // 异常情况readOffset返回-2，这里再返-1
}
break;
}
这里就走到了我们上面提到的readOffset方法了！
readOffset
readOffse方法t的入参是当前分配到的messageQueue和 固定的ReadOffsetType.READ_FROM_STORE，意思就从远程Broker读取该MessageQueue的消费进度。因此走到的是READ_FROM_STORE这个分支，如下所示：
case READ_FROM_STORE: {
try {
long brokerOffset = this.fetchConsumeOffsetFromBroker(mq);// 从broker获取消费进度
AtomicLong offset = new AtomicLong(brokerOffset);
this.updateOffset(mq, offset.get(), false); // 更新消费进度，更新的是RemoteBrokerOffsetStore.offsetTable这个表
return brokerOffset;
}
// No offset in broker
catch (MQBrokerException e) {
return -1;
}
//Other exceptions
catch (Exception e) {
log.warn("fetchConsumeOffsetFromBroker exception, " + mq, e);
return -2;
}
}
fetchConsumerOffsetFromBroker就是往该MessageQueue所在的broker发送获取消费进度的请求了，底层通讯之前的文章已经讲过了，这里就不在赘述了。在Broker端，消费进度保存在ConsumerOffsetManager里面：
key是Topic@ConsumerGroup，value存的是queueId和offset的映射关系。查找消费进度的代码如下：
long offset =
this.brokerController.getConsumerOffsetManager().queryOffset(
requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());
从key来看，topic的消费进度是按照ConsumerGroup来区分的，不同的ConsumerGroup下该MessageQueue的消费进度互不影响，这一点也很好理解。
消息获取与更新消费进度
到目前为止，我们知道了每个MessageQueue的消费进度存在对应的Broker端，在负载均衡服务对每个Topic做负载均衡的时候，创建了PullRequest，并读取了消费进度offset。然后将PullRequest放入了一个阻塞队列中(pullRequestQueue)，供专门的拉取消息线程服务(PullMessageService)读取，然后发起真正的拉取消息请求。这里更新消费进度与获取消息密切相关，因此会涉及一些获取消息的内容。
PullMessageSevice是一个服务线程，专门用于拉取消息，其run方法如下所示：
@Override
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
PullRequest pullRequest = this.pullRequestQueue.take(); // 从阻塞队列中获取一个PullsRequest
this.pullMessage(pullRequest); // 拉取消息
} catch (InterruptedException ignored) {
} catch (Exception e) {
log.error("Pull Message Service Run Method exception", e);
}
}
log.info(this.getServiceName() + " service end");
}
首先从阻塞队列pullRequestQueue中取出PullRequest，然后就是开始调用pullMessage方法获取消息了。调用最终会走到DefaultMQPushConsumerImpl的pullMessage方法中来，代码很多并且如何拉取消息不是本次重点内容，我们这里只贴最后的发送请求部分，理解的其中的参数，也就明白了消息获取请求。
try {
this.pullAPIWrapper.pullKernelImpl(
pullRequest.getMessageQueue(), //消息队列
subExpression,// 订阅表达式，例如"TAG_A"
subscriptionData.getExpressionType(), // 表达式类型，例如"TAG"
subscriptionData.getSubVersion(), // 版本号
pullRequest.getNextOffset(), // 下个消息进度
this.defaultMQPushConsumer.getPullBatchSize(), // 一次性拉取多少条，默认32
sysFlag,// 一些标志位集合，暂时不关心
commitOffsetValue, //
BROKER_SUSPEND_MAX_TIME_MILLIS,// 长轮询时被hold住时间，默认15秒
CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,// 调用超时时间，默认30秒
CommunicationMode.ASYNC, // 异步通讯
pullCallback// 回调
);
} catch (Exception e) {
log.error("pullKernelImpl exception", e);
this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
}
Consumer端更新消费进度
由于发送消息获取请求是异步操作，返回处理在pullCallback里面，因此我们可以大胆猜测，Consumer端消费进度的更新也肯定在这里面。确实，在pullCallback里面会将PullRequest的nextOffset更新：
broker端更新消费进度
由于消息处理不是本次重点内容，所以Broker端对获取消息的处理，我们不打算深入，仅仅需要知道Broker端获取消息后，会计算出一个nextBeginOffset，它就是下个消费进度，然后会返回到Consumer端去供Consumer端更新进度，如下所示：
nextBeginOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
其次，获取完消息后，broker端也会更新消费进度，如下所示：
boolean storeOffsetEnable = brokerAllowSuspend; // brokerAllowSuspend默认是true，如果没有消息就会hold住请求
storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag; // 拉取消息时如果允许提交消费进度，commitOffsetFlag就有
storeOffsetEnable = storeOffsetEnable
&& this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE;// 所以Broker master节点默认情况下这个是true
if (storeOffsetEnable) {
this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),
requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset()); // 保存消费进度，写入offsetTable
}
消费进度持久化
Broker更新消费进度，仅仅是更新了offsetTable这个表，并没有涉及到ConsumerOffset.json这个文件。其实，在Broker初始化时，会启动一项定时任务，定期保存tableOffset到ConsumerOffset.json文件中，如下所示：
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
BrokerController.this.consumerOffsetManager.persist(); // 保存文件
} catch (Throwable e) {
log.error("schedule persist consumerOffset error.", e);
}
}
}, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
flushConsumerOffsetIntervval默认是5s，也就是每隔5保存一次消费进度到文件中。保存的过程是先将原来的文件存到ConsumerOffset.json.bak文件中，然后将新的内容存入ConsumerOffset.json文件。
至此，ConsumerQueue的消费进度维护就算完成了。
小结
RocketMQ每个Topic的消息，在各自的ConsumerGroup下，每个MessageQueue的消费进度是存在broker端的一个consumerOffset.json文件中。Consumer端启动时，会创建PullRequest请求，此时会向Broker发送获取下个消费进度的请求，Broker读取下个消费进度并返回给Consumer端。然后Consumer通过单独的服务线程读取PullRequest并据此拉取消息，Broker端获取到消息后，会及时更新消费进度。此外还有一个单独的定时任务定期保存消费进度到文件，并备份原文件。
GAMELOFT9
发布了379 篇原创文章 · 获赞 85 · 访问量 59万+
他的留言板
关注
标签：消费,PullRequest,broker,进度,Consumer,RocketMQ,消息 来源： https://blog.csdn.net/GAMEloft9/article/details/103999826