RocketMQ消费端有两种获取消息的方式，Push方式和Pull方式。但这两种方式都有一定的缺陷，后来采用了一种折中的方法，采用”长轮询“的方式，它既可以拥有Pull的优点，又能达到保证实时性的目的。

长轮询的思想：
服务端接收到新消息请求后，如果队列里没有新消息，并不急于返回，通过一个循环不断查看状态，每次waitForRunning一段时间（默认是5秒），然后再Check。Broker默认最长阻塞时间为15秒，默认情况下当Broker一直没有新消息，第三次Check的时候，等待时间超过最长阻塞时间，就返回空结果。在等待的过程中，Broker收到了新的消息后会直接返回请求结果。
“长轮询”的核心是，Broker端hold住客户端过来的请求一小段时间。在这段时间内有新的消息到达，就利用现有的连接立即返回消息给Consumer。

何时调用？

当未在Broker中查找到新信息时，状态代码为PULL_NOT_FOUND,会创建拉取任务PullRequest并提交到PullRequestHoldService线程中。

1
2
private ConcurrentMap<String/* topic@queueId */, ManyPullRequest> pullRequestTable =
        new ConcurrentHashMap<String, ManyPullRequest>(1024);

该类中有一个重要的参数pullRequestTable,key为“主题@队列号”，value是对应的ManyPullRequest。

先看一下它的run方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public void run() {
       log.info("{} service started", this.getServiceName());
       while (!this.isStopped()) {
           try {
               //Consumer订阅消息时，Broker是否开启长轮询
               if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {                //开启长轮询，每5秒尝试一次
                   this.waitForRunning(5 * 1000);
               } else {
                 //没有开启长轮询，默认等待1秒再次尝试                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());
   }

从上面可以看出，如果开启了长轮询，每5s尝试一次，利用checkHoldRequest方法来判断是否有新消息的产生。如果未开启长轮询，则默认1s再次尝试。

然后再阅读一下checkHoldRequest方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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);
                }
            }
        }
    }

从上面可以看出，它会遍历pullRequestTable，从key名中可以得到主题名topic和队列名queueId,然后通过topic和queueID获取到该消息队列的最大偏移量，之后调用notifyMessageArriving方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
public void notifyMessageArriving(final String topic, final int queueId, final long maxOffset, final Long tagsCode,
                                     long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
       String key = this.buildKey(topic, queueId);
       ManyPullRequest mpr = this.pullRequestTable.get(key);
       if (mpr != null) {
           List<PullRequest> requestList = mpr.cloneListAndClear();
           if (requestList != null) {
               List<PullRequest> replayList = new ArrayList<PullRequest>();

               for (PullRequest request : requestList) {
                   long newestOffset = maxOffset;
                   if (newestOffset <= request.getPullFromThisOffset()) {
                       newestOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
                   }

                   if (newestOffset > request.getPullFromThisOffset()) {
                       boolean match = request.getMessageFilter().isMatchedByConsumeQueue(tagsCode,
                           new ConsumeQueueExt.CqExtUnit(tagsCode, msgStoreTime, filterBitMap));
                       // match by bit map, need eval again when properties is not null.
                       if (match && properties != null) {
                           match = request.getMessageFilter().isMatchedByCommitLog(null, properties);
                       }

                       if (match) {
                           try {
                               this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                   request.getRequestCommand());
                           } catch (Throwable e) {
                               log.error("execute request when wakeup failed.", e);
                           }
                           continue;
                       }
                   }

                   if (System.currentTimeMillis() >= (request.getSuspendTimestamp() + request.getTimeoutMillis())) {
                       try {
                           this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                               request.getRequestCommand());
                       } catch (Throwable e) {
                           log.error("execute request when wakeup failed.", e);
                       }
                       continue;
                   }

                   replayList.add(request);
               }

               if (!replayList.isEmpty()) {
                   mpr.addPullRequest(replayList);
               }
           }
       }
   }

在notifyMessageArriving方法中，首先会获取到当前该主题、队列中的所有的挂起拉取任务，如果该消息队列的最大偏移量大于待拉取偏移量，说明有新的消息传入。如果消息匹配后，则调用executeRequestWhenWakeup将消息返回给消息拉取客户端，否则等待下一次尝试。
如果挂起超时时间超时，则不继续等待将直接返回客户消息未找到。