概述
当消息被存储后，消费者就会将其消费。
这句话简要的概述了一条消息的最总去向，也引出了本文将讨论的问题：
-
消息什么时候才对被消费者可见？
是在 page cache 中吗？还是在落盘后？还是像 Kafka 一样维护了一个 ISR 队列，等到副本都将消息也落盘后才可见？
-
消息如何到达消费者手里？
是由 Broker push 过去吗？还是由消费者自己 pull？
怎样知道消息消费到哪里？进度由谁管理？是可靠的吗？
...
本文接下来将从消费者的客户端开始介绍，逐步回答以上问题
Pull
Client
RocketMQ 中，具有三个客户端类：
- DefaultMQPullConsumer（deprecated）
- DefaultLitePullConsumer
- DefaultMQPushConsumer
其中，前两个为 Pull 类型，即由自己去拉取消息；后面一个是 Push 类型，即只需要设置好回调方法等，然后等待消息到来后进行调用该方法即可。
但实际上，他们的底层实现都是 pull，即都是由客户端去 Broker 获取消息。
这是因为，使用 Push 需要额外考虑一些问题，如消费者的消费速率慢于 Broker 的发送速率时会导致消费者的缓冲区满，即使可以通过设置背压（BackPressure）机制来做流量控制，但这样毫无疑问会增加程序的复杂度。但如果是消费者按需拉取的话，则设计方法会简便很多，也可以将消费者缓冲区满的问题转化为 Broker 消息堆积的问题。
我们首先来看第二个客户端类 DefaultLitePullConsumer（第一个已经被 RocketMQ 标记为 deprecated，所以不进行介绍）
该 Pull 客户端有两种使用方式，第一种是 Assign，即由我们自己分配要拉取的 queue
// 获取 Topic 的所有 Queue
Collection<MessageQueue> mqSet = litePullConsumer.fetchMessageQueues("TopicTest");
List<MessageQueue> list = new ArrayList<>(mqSet);
// 将要订阅的 queue 加入到 List
List<MessageQueue> assignList = new ArrayList<>();
for (int i = 0; i < list.size() / 2; i++) {
assignList.add(list.get(i));
}
// 手动为这个消费者分配一个消息队列列表
litePullConsumer.assign(assignList);
/*
* 覆盖消费者对于一个 queue 的消费偏移量
* 如果针对同一个消息队列多次调用此 API，则在下一次 poll() 中将使用最新的偏移量
* 请注意，如果在消费过程中随意使用该API，可能会丢失数据
*/
litePullConsumer.seek(assignList.get(0), 10);
第二种是订阅，我们只需要注册需要拉取的 Topic，对于从哪个 queue 拉取，我们并不关心，由系统自动分配
// 选择从 queue 头部开始 pull 还是从尾部开始 pull
litePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
litePullConsumer.subscribe("TopicTestA", "*");
以下为两种操作的内部实现
public synchronized void assign(Collection<MessageQueue> messageQueues) {
setSubscriptionType(SubscriptionType.ASSIGN);
assignedMessageQueue.updateAssignedMessageQueue(messageQueues);
if (serviceState == ServiceState.RUNNING) {
updateAssignPullTask(messageQueues);
}
}
public synchronized void subscribe(String topic, String subExpression) throws MQClientException {
setSubscriptionType(SubscriptionType.SUBSCRIBE);
SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(topic, subExpression);
this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
this.defaultLitePullConsumer.setMessageQueueListener(new MessageQueueListenerImpl());
assignedMessageQueue.setRebalanceImpl(this.rebalanceImpl);
if (serviceState == ServiceState.RUNNING) {
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
updateTopicSubscribeInfoWhenSubscriptionChanged();
}
}
对于消息队列的内部的分配类 AssignedMessageQueue ，Assign 会直接设置要订阅的 queue，Subscribe 会通过设置 rebalance 服务来自动的更新订阅的 queue。
由此可见，两种模式的区别主要在于 rebalance 服务是否启动。
DefaultLitePullConsumer 客户端有以下重要的属性
public class DefaultLitePullConsumer extends ClientConfig implements LitePullConsumer {
/**
* pull 客户端的实现类，几乎全数操作都交由该类执行
*/
private final DefaultLitePullConsumerImpl defaultLitePullConsumerImpl;
/**
* 属于同一个消费者组的消费者共享一个组 ID。然后，组中的消费者通过建立每个队列
* 仅由组中的单个消费者使用来尽可能公平地划分主题。如果所有的消费者都来自同一个组，
* 它的作用就像一个传统的消息队列。每条消息只能由该组的一个消费者使用。
* 当存在多个消费群体时，数据消费模型的流程与传统的发布订阅模型一致。
* 消息被广播到所有消费者组。
*/
private String consumerGroup;
/**
* 长轮询模式中，消费者连接最大挂起时间。
* 不推荐修改
*/
private long brokerSuspendMaxTimeMillis = 1000 * 20;
/**
* 长轮询模式中，消费者最长的超时时间
* 不推荐修改
*/
private long consumerTimeoutMillisWhenSuspend = 1000 * 30;
/**
* socket 超时时间
*/
private long consumerPullTimeoutMillis = 1000 * 10;
/**
* 消费模式，默认为集群
*/
private MessageModel messageModel = MessageModel.CLUSTERING;
/**
* 当前 Consumer 的消息监听器
*/
private MessageQueueListener messageQueueListener;
/**
* 持久化维护的偏移量
*/
private OffsetStore offsetStore;
/**
* 队列分配策略
* - AllocateMachineRoomNearby，根据机房进行分配
* - AllocateMessageQueueAveragely，均分哈希策略
* - AllocateMessageQueueAveragelyByCircle，环形哈希策略
* - AllocateMessageQueueByConfig，根据配置进行分配
* - AllocateMessageQueueByMachineRoom，机房哈希
* - AllocateMessageQueueConsistentHash，一致性哈希
*
* 默认为均分策略
*/
private AllocateMessageQueueStrategy allocateMessageQueueStrategy = new AllocateMessageQueueAveragely();
/**
* 是否开启自动提交偏移量
*/
private boolean autoCommit = true;
/**
* pull 的线程数量
*/
private int pullThreadNums = 20;
/**
* 最小自动提交的时间间隔(毫秒)
*/
private static final long MIN_AUTOCOMMIT_INTERVAL_MILLIS = 1000;
/**
* 最大自动提交的时间间隔(毫秒)
*/
private long autoCommitIntervalMillis = 5 * 1000;
/**
* 每次拉取的最大消息数
*/
private int pullBatchSize = 10;
/**
* 消费请求的流量控制阈值，每个消费者默认最多缓存 10000 个消费请求。
* 缓存是指：已经拉取到了 Consumer, 但还没被消息完成的消息
* 考虑到 {@code pullBatchSize}，瞬时值可能会超过这个限制
*/
private long pullThresholdForAll = 10000;
/**
* 消费的最大偏移跨度
*/
private int consumeMaxSpan = 2000;
/**
* Queue 级别的流量控制阈值，每个 queue 默认最多缓存 1000 条消息
* 考虑到 {@code pullBatchSize}，瞬时值可能会超过限制
*/
private int pullThresholdForQueue = 1000;
/**
* 在队列级别限制缓存的消息大小，每个 queue 默认最多缓存 100 MiB 消息
* 考虑到 {@code pullBatchSize}，瞬时值可能会超过限制
*
* 消息的大小仅包括消息的 body ，因此不准确
*/
private int pullThresholdSizeForQueue = 100;
/**
* poll 的默认超时时间(毫秒)
*/
private long pollTimeoutMillis = 1000 * 5;
/**
* 检查 Topic 元数据更改的时间间隔
*/
private long topicMetadataCheckIntervalMillis = 30 * 1000;
/**
* 刚进入 queue 时的消费位置，默认从尾部进行消费
*/
private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
/**
* 消费回溯时间(?)(秒精度).
* 默认为半小时前
*/
private String consumeTimestamp = UtilAll.timeMillisToHumanString3(System.currentTimeMillis() - (1000 * 60 * 30));
}
启动
DefaultLitePullConsumer 在 Start 时，会先设置消费者组，然后调用刚刚说过的"内部实现类"的 start() 方法
以下是实际的启动方法：
public synchronized void start() throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST:
this.serviceState = ServiceState.START_FAILED;
this.checkConfig();
if (this.defaultLitePullConsumer.getMessageModel() == MessageModel.CLUSTERING) {
this.defaultLitePullConsumer.changeInstanceNameToPID();
}
// 向 MQClintFactory 注册自己的消费者组
initMQClientFactory();
// 初始化再均衡服务
initRebalanceImpl();
initPullAPIWrapper();
// 初始化内部偏移量持久化服务
initOffsetStore();
mQClientFactory.start();
// 开启调度服务
startScheduleTask();
this.serviceState = ServiceState.RUNNING;
log.info("the consumer [{}] start OK", this.defaultLitePullConsumer.getConsumerGroup());
// 启动后置处理
operateAfterRunning();
break;
case RUNNING:
case START_FAILED:
case SHUTDOWN_ALREADY:
throw new MQClientException("The PullConsumer service state not OK, maybe started once, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
default:
break;
}
}
向全局的 MQClientFactroy 实例注册自己的消费者组和实例
初始化 RebalanceImpl。向其注册 消费者组、消费模式(集群或广播)、队列分配策略
创建消息 pull 执行类
-
初始化并加载消费偏移量持久化类
广播模式下使用
LocalFileOffsetStore类，其会从本地文件加载存储 queue 的消费偏移量集群模式下使用
RemoteBrokerOffsetStore类，其会从 Broker 获取 queue 的消费偏移量
启动可能未启动的
MQClientFactory开启调度服务
private void startScheduleTask() {
// 监听 Topic 下的 Queue 发生变化
scheduledExecutorService.scheduleAtFixedRate(
new Runnable() {
@Override
public void run() {
try {
fetchTopicMessageQueuesAndCompare();
} catch (Exception e) {
log.error("ScheduledTask fetchMessageQueuesAndCompare exception", e);
}
}
}, 1000 * 10, this.getDefaultLitePullConsumer().getTopicMetadataCheckIntervalMillis(), TimeUnit.MILLISECONDS);
}
- 将状态标记为已启动
- 进行后置操作
private void operateAfterRunning() throws MQClientException {
// 如果在启动之前就已经进行了 subscribe，则在初始化后更新 Topic 订阅信息
if (subscriptionType == SubscriptionType.SUBSCRIBE) {
updateTopicSubscribeInfoWhenSubscriptionChanged();
}
// 如果在启动之前就已经进行了 assign ，则在初始化后更新 pull 任务。
if (subscriptionType == SubscriptionType.ASSIGN) {
updateAssignPullTask(assignedMessageQueue.messageQueues());
}
// 获取所有注册了 queue change 监听器的 Topic
// 然后获取 Topic 的 Queue, 并添加到本地
for (String topic : topicMessageQueueChangeListenerMap.keySet()) {
Set<MessageQueue> messageQueues = fetchMessageQueues(topic);
messageQueuesForTopic.put(topic, messageQueues);
}
// 获取并存储所有需要订阅的 Topic 所在的所有 Broker 的地址
this.mQClientFactory.checkClientInBroker();
}
然后，该 Consumer 就初始化完成开始工作了
Poll
DefaultLitePullConsumerImpl#poll 方法的主要逻辑是这样的：
public synchronized List<MessageExt> poll(long timeout) {
// 自动提交选择
if (defaultLitePullConsumer.isAutoCommit()) {
maybeAutoCommit();
}
// 等待消费请求
ConsumeRequest consumeRequest = consumeRequestCache.poll(endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
// 更新内存中的偏移量
if (consumeRequest != null && !consumeRequest.getProcessQueue().isDropped()) {
List<MessageExt> messages = consumeRequest.getMessageExts();
long offset = consumeRequest.getProcessQueue().removeMessage(messages);
assignedMessageQueue.updateConsumeOffset(consumeRequest.getMessageQueue(), offset);
return messages;
}
return Collections.emptyList();
}
可以看出，这个方法主要是等待 consumeRequestCache 这个同步阻塞队列被放入消息。那么是在哪里被放入的呢？
对于 Assign 模式，在上面讲到的消费者的启动中，第 8 步的后置操作的 updateAssignPullTask 方法中，会执行 startPullTask 方法。它为所有被分配的 Queue 都添加一个 Pull 任务
private void startPullTask(Collection<MessageQueue> mqSet) {
for (MessageQueue messageQueue : mqSet) {
if (!this.taskTable.containsKey(messageQueue)) {
PullTaskImpl pullTask = new PullTaskImpl(messageQueue);
this.taskTable.put(messageQueue, pullTask);
this.scheduledThreadPoolExecutor.schedule(pullTask, 0, TimeUnit.MILLISECONDS);
}
}
}
对于 Subscribe 模式，他会由之前注册的默认的 MessageQueueListener 来管理，其通过监听 Queue 的变化来更新 Pull 任务
class MessageQueueListenerImpl implements MessageQueueListener {
@Override
public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
MessageModel messageModel = defaultLitePullConsumer.getMessageModel();
switch (messageModel) {
case BROADCASTING:
updateAssignedMessageQueue(topic, mqAll);
updatePullTask(topic, mqAll);
break;
case CLUSTERING:
updateAssignedMessageQueue(topic, mqDivided);
updatePullTask(topic, mqDivided);
break;
default:
break;
}
}
}
在 updatePullTask 方法中，最后也是调用 startPullTask 来启动任务。
所以最后，poll 实现的重点位置就在于 PullTaskImpl 中
而这块的代码比较长，所以我们跳过上半部的代码，只简单概括下：
上半部主要是对当前 Queue 的状态进行多个判断，当可能发生风险的时候直接添加下一次的延时任务，跳过当前状况。
这里的需要自己添加任务的原因是：该任务调度线程池并不是一个定时触发的线程池，而是一个延时任务线程池，所以需要在一次任务执行完成后再继续添加新的任务
会发生跳过的状况如下：
- Pull 任务已经取消
- 队列长度 * 批量消息最大长度 > 最大缓存请求数量，说明可能超过阈值
- queue 的消息缓存数量超过限制
- queue 的消息缓存大小超过限制
- queue 的消息的最大时间与最小时间的差值，超过了指定了允许的差值
- 获取偏移量失败
当将所有的情况都判断完成后，就开始执行 pull 和 将返回的消息包装成 ConsumeRequest
long pullDelayTimeMills = 0;
try {
// 创建订阅信息
SubscriptionData subscriptionData;
String topic = this.messageQueue.getTopic();
if (subscriptionType == SubscriptionType.SUBSCRIBE) {
subscriptionData = rebalanceImpl.getSubscriptionInner().get(topic);
} else {
subscriptionData = FilterAPI.buildSubscriptionData(topic, SubscriptionData.SUB_ALL);
}
// 根据订阅信息远程 pull 消息
PullResult pullResult = pull(messageQueue, subscriptionData, offset, defaultLitePullConsumer.getPullBatchSize());
if (this.isCancelled() || processQueue.isDropped()) {
return;
}
switch (pullResult.getPullStatus()) {
case FOUND:
// 获取 queue 的锁
final Object objLock = messageQueueLock.fetchLockObject(messageQueue);
synchronized (objLock) {
if (pullResult.getMsgFoundList() != null && !pullResult.getMsgFoundList().isEmpty() && assignedMessageQueue.getSeekOffset(messageQueue) == -1) {
processQueue.putMessage(pullResult.getMsgFoundList());
// 提交消费请求
submitConsumeRequest(new ConsumeRequest(pullResult.getMsgFoundList(), messageQueue, processQueue));
}
}
break;
case OFFSET_ILLEGAL:
log.warn("The pull request offset illegal, {}", pullResult.toString());
break;
default:
break;
}
// 更新偏移量
updatePullOffset(messageQueue, pullResult.getNextBeginOffset(), processQueue);
} catch (Throwable e) {
pullDelayTimeMills = pullTimeDelayMillsWhenException;
log.error("An error occurred in pull message process.", e);
}
if (!this.isCancelled()) {
scheduledThreadPoolExecutor.schedule(this, pullDelayTimeMills, TimeUnit.MILLISECONDS);
} else {
log.warn("The Pull Task is cancelled after doPullTask, {}", messageQueue);
}
在拉取消息，包装为 ConsumeRequest 后，就会投入 consumeRequestCache，然后阻塞在 poll 那边的消息就会被唤醒，然后将其返回给上层应用。这也就回答了我们上面提出的那个问题。
而上面的那块的 pull() 方法，则会使用 PullAPIWrapper 来构造一个 pull 请求，然后交给 MQClientFactory 进入底层的 Netty 组件发送，具体的底层发送流程与逻辑我们已经在以下几篇文章中讨论过了
- RocketMQ源码详解 | Producer篇 · 其一：Start，然后 Send 一条消息
- RocketMQ源码详解 | Producer篇 · 其二：消息组成、发送链路
- RocketMQ源码详解 | Broker篇 · 其一：线程模型与接收链路
Commit
Client 的偏移量的提交实现比较简单，这里只简单进行描述。
在 comsumer 调用 commitSync 函数后，会根据当前的消费模式（广播 or 集群）来做不同的操作
-
集群
此时，会将所有的消费队列的偏移量存储到另一个偏移量管理器，只是保存到内存。
-
广播
广播除了会保存到偏移量管理器外，还会像 Broker 的偏移量管理一样持久化到磁盘。
集群模式下的持久化只会在 shutdown 的时候和 pull 时提交到 Broker
Broker
然后来看 Broker 端，在这边，Broker 将 Pull 消息的请求码注册到了 PullMessageProcessor
该类对于 Pull 请求的处理的主要流程如下：
private RemotingCommand processRequest(final Channel channel, RemotingCommand request, boolean brokerAllowSuspend)
throws RemotingCommandException {
/* pass */
// 取出订阅配置
SubscriptionGroupConfig subscriptionGroupConfig =
this.brokerController.getSubscriptionGroupManager().findSubscriptionGroupConfig(requestHeader.getConsumerGroup());
if (null == subscriptionGroupConfig) {
response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
response.setRemark(String.format("subscription group [%s] does not exist, %s", requestHeader.getConsumerGroup(), FAQUrl.suggestTodo(FAQUrl.SUBSCRIPTION_GROUP_NOT_EXIST)));
return response;
}
/* pass */
// 通过消费者组、Topic、Queue、offset、最大消息数、消息过滤器，从 MessageStore 取出消息
final GetMessageResult getMessageResult =
this.brokerController.getMessageStore().getMessage(requestHeader.getConsumerGroup(),
requestHeader.getTopic(),
requestHeader.getQueueId(),
requestHeader.getQueueOffset(),
requestHeader.getMaxMsgNums(),
messageFilter);
if (getMessageResult != null) {
response.setRemark(getMessageResult.getStatus().name());
responseHeader.setNextBeginOffset(getMessageResult.getNextBeginOffset());
responseHeader.setMinOffset(getMessageResult.getMinOffset());
responseHeader.setMaxOffset(getMessageResult.getMaxOffset());
// 设置建议的拉取 Broker 地址
// 这是因为主 Broker 发生了消息堆积，所以交由从 Broker 去接管读请求
if (getMessageResult.isSuggestPullingFromSlave()) {
responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
} else {
responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
}
switch (this.brokerController.getMessageStoreConfig().getBrokerRole()) {
case ASYNC_MASTER:
case SYNC_MASTER:
break;
case SLAVE:
if (!this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
}
break;
}
// 允许从 salve broker 读的话
if (this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
// 消费进度慢，重定向到另一台机器
if (getMessageResult.isSuggestPullingFromSlave()) {
responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
}
// consume ok
else {
responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
}
} else {
responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
}
}
/* pass */
// 持久化偏移量
boolean storeOffsetEnable = brokerAllowSuspend;
storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag;
storeOffsetEnable = storeOffsetEnable
&& this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE;
if (storeOffsetEnable) {
this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),
requestHeader.getConsumerGroup(),
requestHeader.getTopic(),
requestHeader.getQueueId(),
requestHeader.getCommitOffset());
}
return response;
}
我们主要来看 取出消息 和 持久化偏移量 的实现
取出消息
取出消息的主要流程如下
int i = 0;
// 最大扫描消息大小为 16000 与 最大允许消息数量*20 的最大值
final int maxFilterMessageCount = Math.max(16000, maxMsgNums * ConsumeQueue.CQ_STORE_UNIT_SIZE);
final boolean diskFallRecorded = this.messageStoreConfig.isDiskFallRecorded();
ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
for (; i < bufferConsumeQueue.getSize() && i < maxFilterMessageCount; i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
// 先从 ConsumerQueue 获取需要消费的消息索引
long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
int sizePy = bufferConsumeQueue.getByteBuffer().getInt();
long tagsCode = bufferConsumeQueue.getByteBuffer().getLong();
maxPhyOffsetPulling = offsetPy;
if (nextPhyFileStartOffset != Long.MIN_VALUE) {
if (offsetPy < nextPhyFileStartOffset)
continue;
}
boolean isInDisk = checkInDiskByCommitOffset(offsetPy, maxOffsetPy);
// 检查消息的总大小是否到达上限
if (this.isTheBatchFull(sizePy, maxMsgNums, getResult.getBufferTotalSize(), getResult.getMessageCount(),
isInDisk)) {
// 达到上限后即可直接返回
break;
}
/* pass */
// 消息过滤
if (messageFilter != null
&& !messageFilter.isMatchedByConsumeQueue(isTagsCodeLegal ? tagsCode : null, extRet ? cqExtUnit : null)) {
if (getResult.getBufferTotalSize() == 0) {
status = GetMessageStatus.NO_MATCHED_MESSAGE;
}
continue;
}
// 从 CommitLog 获取消息
SelectMappedBufferResult selectResult = this.commitLog.getMessage(offsetPy, sizePy);
if (null == selectResult) {
if (getResult.getBufferTotalSize() == 0) {
status = GetMessageStatus.MESSAGE_WAS_REMOVING;
}
nextPhyFileStartOffset = this.commitLog.rollNextFile(offsetPy);
continue;
}
/* pass */
this.storeStatsService.getGetMessageTransferedMsgCount().incrementAndGet();
// 添加到结果集
getResult.addMessage(selectResult);
status = GetMessageStatus.FOUND;
nextPhyFileStartOffset = Long.MIN_VALUE;
}
/* pass */
// 当消费进度落后物理内存的 40% 时，调换到从库去处理读
long diff = maxOffsetPy - maxPhyOffsetPulling;
long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
* (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
getResult.setSuggestPullingFromSlave(diff > memory);
其中我们可以发现，我们必须要先从 ConsumerQueue 取出对应的消息，然后才进行拉取消息。
而我们在上一章讨论过，ConsumerQueue 由 CommitLogDispatcher 分发后进行维护，即最快可以还在内存中时就可以构建出索引。
这便回答了我们在开头提出的第一个问题：
- 如果没有同步的刷盘策略和副本同步策略的话，我们很快就能对其进行消费（即使消息还在 Page Cache 中）
- 如果存在以上两种策略任一的话，则需要先完成设定的策略，然后才能消费
持久化偏移量
然后是持久化偏移量，消费进度的管理由 Broker 内部维护，但在 Consumer 本地也会有进行管理，且以消费者的消费进度为主
这是因为我们在上面看到的一个特性所导致的，即在消费进度落后物理内存 40% 的时候，会交由从库去读。
这样的特性导致了在主 Broker 中维护的偏移量发生了延迟，即使 salve broker 会通过定期上报偏移量的方法来维护，但难免存在落后。
偏移量的维护比较简单，在 ConsumerOffsetManager 类内部具有一个 并发安全的 Map 来保存
private void commitOffset(final String clientHost, final String key, final int queueId, final long offset) {
ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);
if (null == map) {
map = new ConcurrentHashMap<Integer, Long>(32);
map.put(queueId, offset);
this.offsetTable.put(key, map);
} else {
Long storeOffset = map.put(queueId, offset);
if (storeOffset != null && offset < storeOffset) {
log.warn("[NOTIFYME]update consumer offset less than store. clientHost={}, key={}, queueId={}, requestOffset={}, storeOffset={}", clientHost, key, queueId, offset, storeOffset);
}
}
}
而持久化管理则是在其继承的 ConfigManager 抽象类中实现的
public synchronized void persist() {
String jsonString = this.encode(true);
if (jsonString != null) {
String fileName = this.configFilePath();
try {
MixAll.string2File(jsonString, fileName);
} catch (IOException e) {
log.error("persist file " + fileName + " exception", e);
}
}
}
且为了保证修改时发生宕机后不会错误，其是在备份源文件后再进行写入的
public static void string2File(final String str, final String fileName) throws IOException {
String tmpFile = fileName + ".tmp";
string2FileNotSafe(str, tmpFile);
String bakFile = fileName + ".bak";
String prevContent = file2String(fileName);
if (prevContent != null) {
string2FileNotSafe(prevContent, bakFile);
}
File file = new File(fileName);
file.delete();
file = new File(tmpFile);
file.renameTo(new File(fileName));
}
最后持久化完成后，则可以在你的本地的持久化目录中，中找到一个 json 文件，打开后就是下面这样
{
"offsetTable":{
"%RETRY%my-consumer@my-consumer":{0:0},
"RMQ_SYS_TRANS_HALF_TOPIC@CID_RMQ_SYS_TRANS":{0:22},
"RMQ_SYS_TRANS_OP_HALF_TOPIC@CID_RMQ_SYS_TRANS":{0:4},
"TopicTestA@lite_pull_consumer_test":{0:0}
}
}
可以看到有我们自己创建的测试 Topic，和系统创建的重试 Topic 与事务有关 Topic
Push
Client
然后进入到第三个客户端类 DefaultMQPushConsumer
对于这个客户端类，其与上一个的区别在于，它注册了一个 MessageListener ，且会在我们在之前看到的 start 中执行并发消费和顺序消费的选择。
// 消息的并行消费和顺序消费
if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
this.consumeOrderly = true;
this.consumeMessageService =
new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
} else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
this.consumeOrderly = false;
this.consumeMessageService =
new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
}
并行消费和顺序消费的区别只在于，顺序消费会对消息所在 Queue 加锁，等待上一批消息消费完成后，才会消费下一批消息；并行消费则不会加锁
我们注册的消息监听器则会被 MQClintInstance 在初始化时启动的 PullMessageService 服务所调用。
@Override
public void run() {
while (!this.isStopped()) {
try {
PullRequest pullRequest = this.pullRequestQueue.take();
this.pullMessage(pullRequest);
} catch (InterruptedException ignored) {
} catch (Exception e) {
log.error("Pull Message Service Run Method exception", e);
}
}
}
上面的 pullMessage() 方法就是 Push 实现的重点了(也很长)
该方法首先会根据当前度量标准判断是否需要延迟拉取。延迟拉取的实现是，设置定时器将新的 PullRequest 在一段时间后重新投入 pullRequestQueue
具体的判断方法就是我们在 Pull 已经讲过的：
- Pull 任务已经取消
- ...
等状况
pull 和 push 在这点的逻辑上是相同的
然后开始构建回调函数，该回调函数会在拉取消息后被调用，其主要做的是错误处理和提交消费请求
下面这段就是在回调函数中的"提交消费请求"的方法
// 提交消费请求
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
pullResult.getMsgFoundList(),
processQueue,
pullRequest.getMessageQueue(),
dispatchToConsume);
该方法根据消费是顺序的还是并发的来做具体的实现
构建发送消息所需要的环境后，便和 Pull 一样调用底层 Netty 组件发送
Push 模式构建的请求与 Pull 模式下构建的请求，区别几乎只在这一段
// Pull 模式下构建的请求
PullSysFlag.buildSysFlag(false, block, true, false);
// Push 模式下构建的请求
PullSysFlag.buildSysFlag(
commitOffsetEnable, // commitOffset
true, // suspend
subExpression != null, // subscription
classFilter // class filter
);
可以看出，Pull 不会提交偏移量，且可能会 suspend，同时使用过滤表达式但不支持过滤器类模式
且由于发送的请求码都是一致的，所以我们可以确定，Pull 和 Push 的实质上的区别在于 Push 会使用长轮询
Broker
由于请求码一样，所以在 Broker 端的处理也和上面讲过的代码一样。
在 Broker 对于 Push 的处理中，如果拉取的消息达到了要求的数量的话，则会直接返回，否则会进入到以下代码
实际上，正如我们刚刚说的，Broker 并不能区分 Push 请求和 Pull 请求，它只是根据 SysFlags 上是否有 suspend 标识来选择是否进入以下代码块，而 Pull 请求也是可以使用这个标识的
if (brokerAllowSuspend && hasSuspendFlag) {
long pollingTimeMills = suspendTimeoutMillisLong;
// 开启长轮询则使用，否则为短轮询
if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {
pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();
}
String topic = requestHeader.getTopic();
long offset = requestHeader.getQueueOffset();
int queueId = requestHeader.getQueueId();
// 构建 PullRequest
PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills,
this.brokerController.getMessageStore().now(), offset, subscriptionData, messageFilter);
this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
response = null;
break;
}
该块会构建 PullRequest 然后等待 Broker 的获取
Broker 的 存储组件层 会在将 reputed 指针推进时获取 PullReqeust 并进行处理
reputed 指针的介绍与推进 可以看这篇文章：RocketMQ源码详解 | Broker篇 · 其三：CommitLog、索引、消费队列