• 面试官:RocketMQ 长轮询是怎么实现的?

面试官:RocketMQ 长轮询是怎么实现的?

2025-04-27 10:40:36 栏目:宝塔面板 66 阅读

大家好,我是君哥。

我们知道,消息队列消费端获取消息的方式包括推模式和拉模式,RocketMQ 并没有实现推模式,RocketMQ 的推模式本质上也是拉模式。他们在实现上有下面的不同:

  • 拉模式需要开发在代码里调用拉取消息的方法,拉取到消息后直接进行消息处理;
  • 推模式是消费者客户端初始化时利用重平衡线程去拉取消息,拉取消息的方法会注册回调函数,拉取到消息后,由回调函数触发监听器(定义处理逻辑)进行消息处理。

RocketMQ 为了提供拉取消息的效率,采用了长轮询机制,避免消费端无效的轮询请求。当消费者发送长轮询请求后,如果 Broker 上没有新消息,则不会立刻返回,而是挂起请求,等待新消息到来或者请求超时。

今天来聊一聊 RocketMQ 的长轮询是怎么实现的。

1 长轮询

长轮询的流程如下图:

图片

客户端建立连接后,发送消息拉取请求,如果服务端有新消息,则返回消息。如果服务端没有新消息,则挂起连接,等待新消息到来后给客户端返回。客户端如果连接超时,则断开连接。

2 RocketMQ 实现

2.1 消费端

RocketMQ 消费端长轮询有 2 个超时设置:

  • brokerSuspendMaxTimeMillis:长轮询,Consumer 拉消息请求在 Broker 挂起超过这个时间,就会给消费端返回响应,无论有没有新消息,单位毫秒。这个参数消费端发送拉取请求时会发给 Broker,Broker 用来判断这个长连接是否超时。
  • consumerTimeoutMillisWhenSuspend:消费端发送拉取请求的超时时间,这个时间要大于 brokerSuspendMaxTimeMillis,客户端初始化时会有校验。

注意,这 2 个超时时间官方都不推荐修改。

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

RocketMQ 在 Broker 端通过设置 longPollingEnable 来开启长轮询,默认是开启。

Broker 长轮询挂起时间使用 suspendTimeoutMillis 来进行控制,前面提到过,这个时间由消费者发送的 brokerSuspendMaxTimeMillis 参数来赋值。

2.2.1 挂起消息

Broker 收到客户端拉取消息请求后,如果没有新消息,则将请求挂起,也就是将请求放到 pullRequestTable。

//PullMessageProcessor#processRequest
case ResponseCode.PULL_NOT_FOUND:

if (brokerAllowSuspend && hasSuspendFlag) {
//suspendTimeoutMillisLong 这个参数就是消费端发来的 consumerTimeoutMillisWhenSuspend
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);
//这里挂起消息
this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);
  response = null;
break;
 }

上面的 suspendPullRequest 调用了 PullRequestHoldService#suspendPullRequest,将请求保存在 pullRequestTable。

2.2.2 处理挂起

消息挂起后,后面怎么恢复呢?这里总需要一个线程去循环处理挂起的消息,这个处理逻辑也在 PullRequestHoldService,看下面代码:

public void run() {
 log.info("{} service started", this.getServiceName());
while (!this.isStopped()) {
try {
   //长轮询模式,等待 5s 后处理
   if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
    this.waitForRunning(5 * 1000);
   } //...
   //这里处理被挂起的请求
   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]);
   finallong 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 方法逻辑如下:

  1. 如果当前请求有新消息到来,则给消费者返回响应;
  2. 如果当前请求没有新消息,但是挂起请求已经超时,则给消费者返回响应;
  3. 否则, 继续挂起,等待 5s 后重复执行上面逻辑。

3 总结

长轮询可以降低无效的轮询请求,提升请求效率。RocketMQ 消费者长轮询支持配置,当消息量不太大,消费者没有必要频繁地请求,这时可以设置成长轮询机制。需要注意的是,消费端设置的请求超时时间必须大于 Broker 轮询时间。


本文地址:https://www.yitenyun.com/122.html

搜索文章

Tags

数据库 API FastAPI Calcite 电商系统 MySQL Web 应用 异步数据库 数据同步 ACK 双主架构 循环复制 Deepseek 宝塔面板 Linux宝塔 Docker 生命周期 JumpServer JumpServer安装 堡垒机安装 Linux安装JumpServer 序列 核心机制 esxi esxi6 root密码不对 无法登录 web无法登录 Windows Windows server net3.5 .NET 安装出错 宝塔面板打不开 宝塔面板无法访问 SSL 堡垒机 跳板机 HTTPS 查看硬件 Linux查看硬件 Linux查看CPU Linux查看内存 无法访问宝塔面板 Windows宝塔 Mysql重置密码 HTTPS加密 连接控制 机制 ES 协同 Oracle 处理机制 Serverless 无服务器 语言 OB 单机版 scp Linux的scp怎么用 scp上传 scp下载 scp命令 Spring SQL 动态查询 技术 存储 缓存方案 缓存架构 缓存穿透 索引 运维 日志文件 MIXED 3 分页查询 架构 InnoDB Rsync 响应模型 修改DNS Centos7如何修改DNS RocketMQ 长轮询 配置 监控 Redis 电商 系统 HexHub 异步化 防火墙 服务器 黑客 Linux 安全 聚簇 非聚簇 MySQL 9.3 sftp 服务器 参数 查询 数据 主库 自定义序列化 group by SQLark PostgreSQL 管理口 数据库锁 开源 存储引擎 线上 库存 预扣 业务 SQLite-Web SQLite 数据库管理工具 Doris SeaTunnel MVCC 人工智能 向量数据库 推荐系统 R edis 线程 共享锁 PG DBA 网络架构 工具 网络配置 加密 场景 流量 数据备份 B+Tree ID 字段 • 索引 • 数据库 AI 助手 Ftp redo log 重做日志 ​Redis 机器学习 推荐模型 信息化 智能运维 Python 优化 万能公式 RDB AOF Redis 8.0 缓存 高可用 Canal GreatSQL 连接数 INSERT COMPACT 核心架构 订阅机制 网络故障 Postgres OTel Iceberg 云原生 Hash 字段 同城 双活 IT运维 微软 SQL Server AI功能 不宕机 自动重启 Web prometheus Alert SVM Embedding 高效统计 今天这篇文章就跟大家 Netstat Linux 服务器 端口 引擎 性能 大模型 虚拟服务器 虚拟机 内存 向量库 Milvus OAuth2 Token Entity 开发 分库 分表 DBMS 管理系统 MongoDB 容器 sqlmock 崖山 新版本 ZODB Undo Log 单点故障 LRU mini-redis INCR指令 JOIN 数据集成工具 容器化 分布式 集中式 悲观锁 乐观锁 openHalo 窗口 函数 SpringAI 排行榜 排序 磁盘架构 QPS 高并发 Testcloud 云端自动化 Redka 数据脱敏 加密算法 Redisson 锁芯 PostGIS 启动故障 大表 业务场景 EasyExcel MySQL8 Recursive 聚簇索引 非聚簇索引 数据类型 StarRocks 数据仓库 分页 数据结构 意向锁 记录锁 Pottery 模型 原子性 分布式架构 分布式锁​ R2DBC AIOPS InfluxDB 1 事务 Java SSH IT Caffeine CP 网络 MCP 开放协议 dbt 数据转换工具 RAG HelixDB 工具链 发件箱模式 池化技术 连接池 Web 接口 部署 filelock 数据分类 字典 传统数据库 向量化 对象 速度 服务器中毒 读写 Go 数据库迁移 事务隔离 双引擎 频繁 Codis 分页方案 排版 数据页 优化器 单线程 仪表盘 Order 线程安全 LLM Crash 代码 订单 List 类型 UUIDv7 主键 日志 Pump 事务同步 Ansible