Understanding RocketMQ Long‑Polling Mechanism and Its Implementation

We know that message‑queue consumers can obtain messages via push or pull modes; RocketMQ does not implement a true push mode—its so‑called push mode is essentially pull, differing mainly in how the client registers callbacks.

To improve pull efficiency, RocketMQ adopts a long‑polling mechanism: when a consumer sends a pull request and the broker has no new messages, the request is suspended instead of returning immediately, waiting for new messages or a timeout.

1. Long Polling

The long‑polling flow is illustrated in the diagram below (image omitted).

After the client establishes a connection and sends a pull request, the server returns messages if available; otherwise it suspends the connection until new messages arrive or the request times out, after which the client disconnects.

2. RocketMQ Implementation

2.1 Consumer Side

RocketMQ’s consumer long‑polling has two timeout configurations:

brokerSuspendMaxTimeMillis : the maximum time the broker will keep a suspended pull request before responding, regardless of whether new messages exist.

consumerTimeoutMillisWhenSuspend : the timeout for the consumer’s pull request, which must be greater than brokerSuspendMaxTimeMillis (validated on client initialization).

Note: both parameters are not recommended to be modified in production.

if (this.defaultLitePullConsumer.getConsumerTimeoutMillisWhenSuspend() < this.defaultLitePullConsumer.getBrokerSuspendMaxTimeMillis()) {
    throw new MQClientException(
        "Long polling mode, the consumer consumerTimeoutMillisWhenSuspend must greater than brokerSuspendMaxTimeMillis" +
        FAQUrl.suggestTodo(FAQUrl.CLIENT_PARAMETER_CHECK_URL),
        null);
}

2.2 Broker Side

The broker enables long polling via the longPollingEnable flag (enabled by default). The suspension duration is controlled by suspendTimeoutMillis , whose value comes from the consumer’s brokerSuspendMaxTimeMillis parameter.

2.2.1 Suspending Requests

When the broker receives a pull request with no new messages, it places the request into pullRequestTable for later processing.

// PullMessageProcessor#processRequest
case ResponseCode.PULL_NOT_FOUND:
    if (brokerAllowSuspend && hasSuspendFlag) {
        // suspendTimeoutMillisLong is the consumerTimeoutMillisWhenSuspend value
        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 = new PullRequest(request, channel, pollingTimeMills,
            this.brokerController.getMessageStore().now(), offset, subscriptionData, messageFilter);
        // suspend the request
        this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
        response = null;
        break;
    }

2.2.2 Processing Suspended Requests

A dedicated thread periodically checks and resumes suspended pull requests. The core loop is shown below:

public void run() {
    log.info("{} service started", this.getServiceName());
    while (!this.isStopped()) {
        try {
            // long‑polling mode, wait 5 seconds before processing
            if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                this.waitForRunning(5 * 1000);
            }
            // handle suspended requests
            this.checkHoldRequest();
        } catch (Throwable e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
        }
    }
}

protected 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]);
            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);
            }
        }
    }
}

The notifyMessageArriving method follows three rules: (1) if new messages arrive, respond to the consumer; (2) if the request has timed out, respond even without new messages; (3) otherwise keep the request suspended and retry after 5 seconds.

3. Summary

Long polling reduces unnecessary poll requests and improves request efficiency. RocketMQ’s consumer long‑polling can be configured, and when message volume is low, consumers need not poll frequently; however, the consumer timeout must always be greater than the broker’s polling timeout.