TDengine 数据订阅支持 MQTT 协议用户手册
TDengine 数据订阅支持 MQTT 协议用户手册
1. 概述
TDengine 3.3.7.0 新增对 MQTT v5.0 协议的支持，允许用户通过 MQTT 客户端直接订阅数据库中的实时数据。该功能通过 Broker Node (bnode) 实现，提供零代码集成能力，显著降低系统复杂度，同时支持数据持久化、高可用和共享订阅等特性。
2. 准备工作
2.1 创建 Broker Node (bnode)
在 TDengine 集群中执行以下命令创建 bnode（需替换 dnode_id 为实际节点 ID）：
CREATE BNODE ON DNODE <dnode_id> PROTOCOL 'mqtt';
- 每个 dnode 仅支持创建一个 bnode
- bnode 创建后自动启动 taosmqtt 进程，默认监听端口 6083
2.2 配置 MQTT 服务
在 taos.cfg 中调整参数：
mqttPort 6083 # MQTT 服务端口（重启生效）
mqtt 1 # 启用 MQTT 服务（0：关闭，1：启用）
3 创建订阅主题
所有主题需预先通过 SQL 定义， 不支持通配符或动态创建 ：
CREATE TOPIC topic_name AS SELECT * FROM super_table;
- 支持订阅库、超级表或查询结果
- 主题不存在时，MQTT 订阅请求将返回失败
使用 MQTT 客户端订阅数据
4.1 连接与认证
客户端连接需提供 TDengine 用户名和密码：
client.connect("bnode_ip", port=6083, username="td_user", password="td_password")
- 仅支持基础身份验证，暂不支持 TLS/SSL
4.2 订阅主题
发送 SUBSCRIBE 请求并指定目标主题：
client.subscribe("topic_name", qos=1)
- QoS 支持 ：
-
qos=0：至多一次（可能丢失数据，适用于实时监控）
-
qos=1：至少一次（数据不丢失，可能重复，适用于告警系统）
-
4.3 消费消息
消息格式通过 MQTT 属性指定：
JSON 格式 （默认）：
{"topic":"topic_name", "rows":[{"col1":value1, "col2":value2}]}
- 二进制格式 ：在订阅请求中添加用户属性 proto: rawblock
5. 高级特性
5.1 共享订阅
通过 $share/ 前缀实现负载均衡：
client.subscribe("$share/group1/topic_name")
-
同一消费组内多个客户端均匀接收数据
-
支持高可用：单个 bnode 宕机不影响服务
5.2 指定消费起始位置
在订阅请求中添加用户属性：
properties = [("sub-offset", "earliest")] # 从最早数据开始消费
client.subscribe("topic_name", properties=properties)
- 默认从最新数据（latest）开始消费
5.3 消息格式控制
通过 MQTT 属性设置内容类型：
properties = [("Content-Type", "TDengineJsonV1.0")] # 显式指定 JSON 格式
6 配置参数详解
| 参数 | 说明 | 默认值 | 动态修改 |
|---|---|---|---|
| mqtt | 启用 MQTT 服务（0/1） | 0 | 重启生效 |
| mqttPort | 监听端口 | 6083 | 否 |
7 常见问题
- Q1: 订阅时提示主题不存在
原因：主题未预先创建。
解决：执行 CREATE TOPIC 定义主题后重试。 - Q2: 共享订阅的消息分配不均
原因：消费组内客户端数量变化触发再平衡。
解决：等待 mnode 自动完成负载分配。 - Q3: QoS 2 是否支持？
暂不支持，QoS 2 请求会降级为 QoS 1
关于 TDengine
TDengine 专为物联网IoT平台、工业大数据平台设计。其中，TDengine TSDB 是一款高性能、分布式的时序数据库（Time Series Database），同时它还带有内建的缓存、流式计算、数据订阅等系统功能；TDengine IDMP 是一款AI原生工业数据管理平台，它通过树状层次结构建立数据目录，对数据进行标准化、情景化，并通过 AI 提供实时分析、可视化、事件管理与报警等功能。