RocketMQ MQTT：面向物联网与移动终端的新一代消息协议架构

技术文档

🚀 RocketMQ MQTT：面向物联网与移动终端的新一代消息 协议架构

RocketMQ MQTT 是 Apache RocketMQ 在 5.0+ 版本中引入的 原生 MQTT 协议支持能力，通过集成 MQTT 协议栈，使 RocketMQ 能够直接接入 物联网（IoT）设备、移动 APP、车联网终端 等轻量级客户端，无需额外网关或桥接服务。

这标志着 RocketMQ 从“传统消息中间件”向 “云边端一体化消息平台” 的演进。

一、为什么需要 RocketMQ MQTT？

在物联网、移动互联网场景下，终端设备（如传感器、手机 APP）普遍使用 MQTT 协议，原因如下：

需求 MQTT 的优势 低带宽 报文精简，头部最小仅 2 字节 低功耗 支持 QoS 0/1/2，节省电量 弱网络适应 支持断线重连、遗嘱消息（Will Message） 海量连接 单节点支持数万级 TCP 连接

而传统 RocketMQ 只支持自研协议，无法直接接入这些终端。

✅ RocketMQ MQTT 的目标：让 RocketMQ 成为“云边端”统一消息中枢。

二、RocketMQ MQTT 架构模型

+------------------+ +------------------+| IoT Devices | | Mobile APPs || (MQTT Clients) | | (MQTT Clients) |+--------+----------+ +--------+----------+ | | +------------+--------------+↓ +--------v---------+ | RocketMQ MQTT | ← 协议解析、会话管理 | (Broker) | +--------+---------+↓ +--------v---------+ | RocketMQ Core | ← CommitLog、ConsumeQueue | (存储与路由) | +--------+---------+↓ +--------v---------+ | Flink / Spark | | (实时计算) | +------------------+

🔧 核心组件：

组件说明 MQTT Broker 内置于 RocketMQ Broker 中，处理 MQTT 连接与消息 Session Manager 管理客户端会话（Clean Session、持久化会话） Subscription Manager 管理主题订阅关系（支持通配符 + 和 #） QoS 处理 支持 QoS 0、1、2 的消息传递语义 Retained Message 保留消息，新订阅者立即收到最新状态 Will Message 遗嘱消息，客户端异常断开时触发

三、核心功能详解

1. MQTT 协议兼容

RocketMQ MQTT 完全兼容 MQTT 3.1.1 和 MQTT 5.0 协议，支持：

特性说明 ✅ CONNECT / PUBLISH / SUBSCRIBE 基础报文 ✅ QoS 0/1/2 最多一次、至少一次、精确一次 ✅ Clean Session 会话清理控制 ✅ Last Will and Testament (LWT) 遗嘱消息 ✅ Retained Messages 保留消息 ✅ Topic Wildcards +（单层通配）、#（多层通配） ✅ Keep Alive 心跳保活 ✅ Session Persistence 持久化会话，支持离线消息

2. 与 RocketMQ 原生能力融合

MQTT 消息最终写入 RocketMQ 的 CommitLog，与其他消息统一存储。

MQTT PUBLISH → RocketMQ MQTT Broker ↓转换为 RocketMQ Message ↓写入 CommitLog（所有 Topic 共用） ↓消费者可通过 RocketMQ SDK 消费

✅ 实现“MQTT 消息 → RocketMQ 消费”的无缝衔接。

3. 消息路由与 Topic 映射

MQTT Topic 与 RocketMQ Topic 直接映射
支持命名空间隔离：tenant/device/status

// MQTT 客户端发布client.publish(\"device/123/status\", \"online\", QoS1);// RocketMQ 消费者订阅consumer.subscribe(\"device/123/status\", \"*\");

4. 离线消息与持久化

支持 持久化会话（Clean Session = false）
客户端离线期间的消息将存储在 ConsumeQueue
上线后自动接收积压消息

✅ 保证 QoS 1/2 的“至少一次”语义。

5. 安全与认证

支持 MQTT 原生 username/password 认证
集成 RocketMQ ACL，控制 Topic 权限
支持 TLS 加密通信

// MQTT 客户端连接MqttConnectOptions options = new MqttConnectOptions();options.setUserName(\"device_123\");options.setPassword(\"secret_key\".toCharArray());options.setSocketFactory(SSLSocketFactory.getDefault());

四、工作流程

1. MQTT 客户端连接 RocketMQ Broker ↓2. 发送 CONNECT 报文（含 ClientId、CleanSession、Will） ↓3. Broker 认证并建立会话 ↓4. 客户端 PUBLISH 消息 ↓5. Broker 将 MQTT 消息转换为 RocketMQ Message ↓6. 写入 CommitLog 并通知订阅者 ↓7. 其他 MQTT 或 RocketMQ 消费者接收消息

五、部署方式

1. 内置模式（推荐）

MQTT 协议栈直接集成在 RocketMQ Broker 中
通过配置启用

# broker.confenableMQTT=truemqttPort=1883mqttSSLPort=8883

✅ 简单高效，适合大多数场景。

2. Proxy 模式（云原生）

使用 RocketMQ Proxy 作为 MQTT 入口
Proxy 解析 MQTT 协议，转发给后端 Broker

MQTT Client → Proxy (1883) → Broker (10911)

✅ 支持多协议统一接入，适合 Kubernetes 环境。

六、使用场景

场景说明 物联网（IoT） 传感器数据上报、设备远程控制 移动 APP 推送 消息通知、状态同步 车联网 车辆状态上报、远程诊断 智能家居 设备联动、状态同步 工业互联网 PLC 数据采集、告警通知

七、与传统 MQTT Broker 对比

特性传统 MQTT Broker（如 Mosquitto） RocketMQ MQTT 存储能力 内存为主，持久化弱 CommitLog 持久化，高可靠 吞吐量 中等百万级 TPS 扩展性 一般支持分片、DLedger 高可用 生态集成 弱可对接 Flink、Streams、Connect 运维监控 简单 Prometheus + Grafana 全链路监控 适用场景 小规模设备接入大规模云边端协同

✅ RocketMQ MQTT = 高性能 + 高可靠 + 云原生

八、最佳实践建议

实践说明 ✅ 合理设计 Topic 层级如 tenant/device/type/id ✅ 控制消息大小建议 < 1MB，避免影响性能 ✅ 使用 QoS 1 平衡可靠与性能 QoS 2 性能开销大 ✅ 启用 TLS 加密保障传输安全 ✅ 监控连接数与消息量防止资源耗尽 ✅ 结合 RocketMQ Console 管理查看客户端连接状态

✅ 总结

维度说明定位 RocketMQ 的“边缘接入层” 核心能力 原生 MQTT 协议支持、与 RocketMQ 存储融合优势高吞吐、高可靠、云边端一体 部署方式 内置 Broker 或通过 Proxy 适用场景 IoT、移动 APP、车联网

🚀 一句话总结：
RocketMQ MQTT 是 RocketMQ 的“终端神经末梢”，
它让 RocketMQ 不仅能处理“服务间消息”，
还能直接接入“设备与用户”，
构建出 从终端到云端的全链路消息闭环。

掌握 RocketMQ MQTT，你就能构建真正意义上的 万物互联、实时响应 的智能系统。

RocketMQ MQTT：面向物联网与移动终端的新一代消息协议架构