【mq读书笔记】消息拉取长轮训机制(Broker端)

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

RocketMQ并没有真正实现推模式，而是消费者主动想消息服务器拉取消息，推模式是循环向消息服务端发送消息拉取请求。

如果消息消费者向RocketMQ发送消息拉取时，消息未到达消费队列：

如果不启用长轮询机制消息并未达到消费队列，则会在服务端等待shortPollingTimeMills时间后再去判断消息是否已到达消息队列。如果消息未到达则提示消息拉取客户端消息不存在；

如果开启长轮训模式，mq一方面会每5s轮询检查一次消息是否可达，同时一有新消息到达后立马通知挂起线程再次验证新消息是否是自己感兴趣的消息，如果是则从commitlog文件提起消息返回给消息拉取客户端，否则直到挂起超时，超时时间由消息拉取方式在消息拉取时封装在请求参数重。push模式默认15s，pull模式通过DefaultMQPullConsumer#setBrokerSuspendMaxTimeMillis设置。

MQ通过在Broker端配置longPollingEnable为true来开启长轮询模式。

消息拉取时“服务端”未找到消息处理逻辑如下：

如果brokerAllowSuspend为true，表示支持挂起，则将响应对象response设置为null，将不会立即想客户端卸乳响应，hasSuspendFlag参数在拉取消息时，候检的拉取标记，默认为true。

这里创建创建拉取任务PullRequest并提交到PullRequestHoldService线程中。

共2个线程共同来完成轮询：

PullRequestHoldService：每个5s重试一次。

DefaultMessageStore.ReputMessageService每处理一次重新拉取，sleep(100)，继续下一次检查。

PullRequestHoldService#suspendPullRequest：

public void suspendPullRequest(final String topic, final int queueId, final PullRequest pullRequest) {
String key = this.buildKey(topic, queueId);
ManyPullRequest mpr = this.pullRequestTable.get(key);
if (null == mpr) {
mpr = new ManyPullRequest();
ManyPullRequest prev = this.pullRequestTable.putIfAbsent(key, mpr);
if (prev != null) {
mpr = prev;
}
} mpr.addPullRequest(pullRequest);
}

ManyPullRequest对象内部持有一个PullRequest列表，表示同一“主题@队列”的累积拉取消息任务。

PullRequestHoldService#run

@Override
public void run() {
log.info("{} service started", this.getServiceName());
while (!this.isStopped()) {
try {
if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
this.waitForRunning(5 * 1000);
} else {
this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
} long beginLockTimestamp = this.systemClock.now();
this.checkHoldRequest();
long costTime = this.systemClock.now() - beginLockTimestamp;
if (costTime > 5 * 1000) {
log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
}
} catch (Throwable e) {
log.warn(this.getServiceName() + " service has exception. ", e);
}
} log.info("{} service end", this.getServiceName());
}

private void checkHoldRequest() {
for (String key : this.pullRequestTable.keySet()) {
String[] kArray = key.split(TOPIC_QUEUEID_SEPARATOR);
if (2 == kArray.length) {
String topic = kArray[0];
int queueId = Integer.parseInt(kArray[1]);
final long offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
try {
this.notifyMessageArriving(topic, queueId, offset);
} catch (Throwable e) {
log.error("check hold request failed. topic={}, queueId={}", topic, queueId, e);
}
}
}
}

遍历拉取任务表，根据主题与队列获取消息消费队列最大偏移量，如果该偏移量大于待拉取偏移量，说明有新的消息到达，调用notifyMessageArriving触发消息拉取。

PullRequestHoldService#notifyMessageArriving(java.lang.String, int, long, java.lang.Long, long, byte[], java.util.Map<java.lang.String,java.lang.String>)：