TDengine 数据订阅（TMQ）完全指南：主题、消费组、Native 与 MQTT 订阅实战-平芜编程栈

数据库
时序数据库
物联网
大数据
实时分析
云原生

【免费下载链接】tdengine

TDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.

项目地址：https://gitcode.com/taosdata/tdengine

点击查看免费下载

TDengine 在时序数据库内核中内置了与 Kafka 高度相似的消息队列能力：用户可以用 SQL 定义主题（topic），下游应用通过多语言连接器或 MQTT 客户端实时消费新写入的数据，从而在监控、告警、实时分析与数据同步等场景中省去额外部署消息队列的成本。本文基于仓库中的数据订阅章节及其配套文档，系统讲解主题语法、消费组与 WAL 存储原理、Native 连接器消费流程以及 MQTT 订阅方案，并给出可直接运行的 SQL、Python 示例与源码级佐证。

为什么 TDengine 需要内置数据订阅

在监控、告警、实时分析和数据同步等场景中，下游程序通常需要第一时间获取新写入的数据。如果通过定时查询拉取数据，不仅延迟更高，也会增加数据库查询压力。TDengine 提供了类似于消息队列产品的数据订阅和消费接口。在许多场景中，采用 TDengine 的时序大数据平台，无须再集成消息队列产品，从而简化应用程序设计并降低运维成本。

与 Kafka 类似，用户需要在 TDengine 中定义主题（topic）。TDengine 的主题可以是一个数据库、一张超级表，或者基于现有超级表、子表或普通表的查询条件，即一条查询语句。用户可以利用 SQL 对标签、表名、列、表达式等条件进行过滤，并对数据进行标量函数与 UDF 计算（不包括数据聚合）。与其他消息队列工具相比，这是 TDengine 数据订阅功能的最大优势：数据的粒度由定义主题的 SQL 决定，过滤与预处理由 TDengine 自动完成，从而减少传输的数据量并降低应用程序的复杂度。

核心概念：主题、消费组与 WAL

消费者与消费组

消费者订阅主题后，可以实时接收最新的数据。多个消费者可以组成一个消费组，共享消费进度，实现多线程、分布式地消费数据，提高消费速度。不同消费组的消费者即使消费同一个主题，也不共享消费进度。一个消费者可以订阅多个主题。如果主题对应的是超级表或库，数据可能会分布在多个不同的节点或数据分片上；当一个消费组中有多个消费者时，可以提高消费效率。

TDengine 的消息队列提供了消息的 ACK（Acknowledgment，确认）机制，确保在宕机、重启等复杂环境下实现至少一次（at least once）消费。

WAL：把预写日志改造为持久化消息存储

为实现上述功能，TDengine 会为预写数据日志（Write-Ahead Logging，WAL）文件自动创建索引，以支持快速随机访问，并提供了灵活可配置的文件切换与保留机制。用户可以根据需求指定 WAL 文件的保留时间和大小。通过这些方法，WAL 被改造成一个保留事件到达顺序的、可持久化的存储引擎。

对于以主题形式创建的查询，TDengine 将从 WAL 读取数据。在消费过程中，TDengine 根据当前消费进度从 WAL 直接读取数据，并使用统一的查询引擎实现过滤、变换等操作，然后将数据推送给消费者。

从源码结构看，WAL 的保留策略在 vnode 层被显式配置与校验。例如 vnodeSvr.c 中处理 vnode 更新请求时会对比walRetentionPeriod、walRetentionSize等字段并写入 vnode 配置；相关配置参数的说明可参考 taosd 配置参数。

主题数量的实例级上限

一个 TDengine 实例可创建的 topic 个数上限由tmqMaxTopicNum参数控制。该参数定义于 tglobal.c（默认值 20，取值范围 1～10000，属全局配置参数，支持通过 SQL 动态修改、立即生效），并在 mnode 侧创建 topic 时被实际校验——mndTopic.c 中通过sdbGetSize(pMnode->pSdb, SDB_TOPIC) >= tmqMaxTopicNum判断是否达到上限。完整参数说明见 taosd 配置参数 · tmqMaxTopicNum。

版本相关行为变更

从v3.2.0.0开始，数据订阅支持 vnode 迁移和分裂。由于数据订阅依赖 WAL 文件，而在 vnode 迁移和分裂过程中，WAL 文件并不会同步。因此，在迁移或分裂操作完成后，将无法继续消费此前尚未消费完的 WAL 数据。请务必在执行 vnode 迁移或分裂之前，将所有 WAL 数据消费完毕。
从v3.3.7.0开始，支持通过 MQTT 客户端订阅已创建主题的数据，详见 MQTT 订阅。

主题类型与创建语法

TDengine 从v3.0.0.0开始对消息队列做了大幅优化和增强，以简化用户的数据订阅方案。用户可以通过 SQL 创建订阅主题，然后使用连接器 API、taosshell 或 MQTT 客户端消费主题中的数据。TDengine 使用 SQL 创建的主题共有 3 种类型。

查询主题（SELECT 主题）

订阅一条 SQL 查询定义的数据流，创建语法如下：

CREATE TOPIC [IF NOT EXISTS] topic_name AS subquery

该 SQL 通过SELECT语句订阅数据，包括SELECT *，或SELECT ts, c1等指定列查询。查询主题可以带条件过滤和标量函数计算，但不支持聚合函数、时间窗口聚合，也不支持DISTINCT、GROUP BY、ORDER BY、PARTITION BY、LIMIT/SLIMIT等。需要注意：

该类型 topic 一旦创建，订阅数据的结构即确定。
被订阅或用于计算的列或标签不可被删除（ALTER TABLE DROP）或修改（ALTER TABLE MODIFY）。从v3.4.0.0开始，可以修改、删除、增加这些列或标签，但需要执行RELOAD TOPIC使变更生效。
对于SELECT *，订阅会展开为创建 topic 时的所有列：子表、普通表为数据列，超级表为数据列加标签列。
不支持虚拟表的查询订阅。
subquery中的超级表、子表、普通表可以被删除。删除后，订阅数据为空；如果删除后重新创建同名表，订阅数据仍然为空，因为表 ID 已变化。如需订阅新建表的数据，可以通过RELOAD TOPIC重新加载 topic。

实战示例：假设需要订阅所有智能电表中电压值大于 200 的数据，且仅仅返回时间戳、电流、电压 3 个采集量（不返回相位），可以通过下面的 SQL 创建power_topic主题：

CREATE TOPIC power_topic AS SELECT ts, current, voltage FROM power.meters WHERE voltage > 200;

超级表主题

订阅一个超级表中的所有数据，语法如下：

CREATE TOPIC [IF NOT EXISTS] topic_name [WITH META | ONLY META] AS STABLE stb_name [where_condition]

与使用SELECT * FROM stbName订阅的区别是：

不会限制用户的表结构变更，即表结构变更以及变更后的新数据都能够订阅到。
返回的是非结构化的数据，返回数据的结构会随着超级表的表结构变化而变化。
WITH META参数可选，指定后将返回创建超级表、子表等语句，主要用于 taosX 做超级表迁移。
ONLY META参数可选，指定后仅订阅元数据变更，不再传输时序数据。
where_condition参数可选，用于过滤符合条件的子表并订阅这些子表。WHERE条件里不能有普通列，只能是 tag 或tbname；可以使用函数过滤 tag，但不能使用聚合函数，因为子表 tag 值无法做聚合；也可以是常量表达式，比如2 > 1（订阅全部子表），或者false（订阅 0 个子表）。
返回数据不包含标签。
支持虚拟超级表的订阅，仅能订阅出虚拟超级表的 meta 信息，所以虚拟超级表订阅需要带上WITH META或ONLY META参数，否则订阅不到内容。

数据库主题

订阅一个数据库里所有数据，语法如下：

CREATE TOPIC [IF NOT EXISTS] topic_name [WITH META | ONLY META] AS DATABASE db_name;

通过该语句可创建一个包含数据库所有表数据的订阅：

WITH META参数可选，指定后将返回数据库里所有超级表、子表、普通表的元数据创建、删除、修改语句，主要用于 taosX 做数据库迁移。
ONLY META参数可选，指定后仅订阅元数据变更，不再传输时序数据。
WITH META或ONLY META的情况下可订阅出虚拟表的信息，并且仅能订阅出虚拟表的 meta 信息。

说明：超级表订阅和库订阅属于高级订阅模式，容易出错，如确实要使用，请咨询技术支持人员。

主题管理：删除、查看与加载

删除主题

如果不再需要订阅数据，可以删除 topic。如果当前 topic 被消费者订阅，通过FORCE语法可强制删除；强制删除后，订阅的消费者在消费数据时会出错（FORCE语法从v3.3.6.0开始支持）。

DROP TOPIC [IF EXISTS] [FORCE] topic_name;

查看主题

SHOW TOPICS;

显示当前数据库下的所有主题信息。更完整字段见元数据表INS_TOPICS。

加载主题（RELOAD TOPIC）

RELOAD TOPIC [IF EXISTS] topic_name AS subquery;

该语法从v3.4.0.0开始支持，仅适用于查询主题，用于重新加载主题定义。它主要解决查询主题里变更列或 tag，以及SELECT *查询订阅时删除或增加列、tag 后输出结果不生效的问题。
需要变更订阅表结构的 schema 时，建议先停止消费，再变更表结构，然后执行RELOAD TOPIC，接着重新开始订阅。

消费者与消费组管理

创建消费者

消费者通常通过 TDengine 客户端驱动或者连接器所提供的 API 创建，详情可以参考开发指南 · 数据订阅。为了快速验证订阅功能，也可以在taosshell 中执行subscribe <topic> -g <group_id>创建消费者并开始消费，具体用法请参考taosCLI 数据订阅。

查看消费者

SHOW CONSUMERS;

显示当前数据库下所有消费者的信息，包括消费者的状态、创建时间等。更完整字段见性能数据表PERF_CONSUMERS。

删除消费组

创建消费者时会为其指定一个消费者组。消费者不能显式地删除，但可以删除消费者组。如果当前消费者组里有消费者在消费，通过FORCE语法可强制删除；强制删除后，订阅的消费者在消费数据时会出错（FORCE语法从v3.3.6.0开始支持）。

DROP CONSUMER GROUP [IF EXISTS] [FORCE] cgroup_name ON topic_name;

查看订阅信息

SHOW SUBSCRIPTIONS;

显示 topic 在不同 vgroup 上的消费信息，可用于查看消费进度。更完整字段见元数据表INS_SUBSCRIPTIONS。

Native 订阅：用连接器 API 消费数据

TDengine 提供了多语言数据订阅 API（包括但不限于创建消费者、订阅主题、取消订阅、拉取数据、提交与设置消费进度等），与 Kafka 订阅 API 保持高度一致，便于复用既有开发经验。目前支持 C、Java、Go、Rust、Python 和 C# 等。各语言连接器对SELECT主题、库/超级表主题以及WITH META/ONLY META消费的支持差异，详见数据订阅编程接口中的能力对照表。

创建主题

如下 SQL 将创建一个名为topic_meters的订阅。使用该订阅所获取的消息中的每条记录都由查询语句SELECT ts, current, voltage, phase, groupid, location FROM meters所选择的列组成。

CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM meters;

创建消费者的常用参数

TDengine 消费者的概念与 Kafka 类似，消费者通过订阅主题来接收数据流。消费者可以配置多种参数，如连接方式、服务器地址、自动提交 Offset、自动重连、数据传输压缩等。常用参数如下：

参数	说明	默认值
`td.connect.ip`	服务端的 FQDN	—
`td.connect.user`	用户名	—
`td.connect.pass`	密码	—
`td.connect.token`	Token（通过`CREATE TOKEN`语句生成；使用 Token 认证时无需设置 user/pass）	—
`td.connect.port`	服务端的端口号	—
`group.id`	消费组 ID，同一消费组共享消费进度（必填项，最大长度 192，不可包含英文冒号`:`；每个 topic 最多可建立 100 个 consumer group）	—
`client.id`	客户端 ID（最大长度 255）	—
`auto.offset.reset`	消费组订阅的初始位置：`earliest`从头订阅（`v3.2.0.0`之前默认）；`latest`仅从最新数据订阅（`v3.2.0.0`及之后默认）；`none`没有已提交 offset 时无法订阅	`latest`
`enable.auto.commit`	是否启用消费位点自动提交（false 时需应用自行 commit）	`true`
`auto.commit.interval.ms`	自动提交消费位点的时间间隔（毫秒）	`5000`
`msg.with.table.name`	是否允许从消息中解析表名（不适用于列订阅，自`v3.2.0.0`起废弃）	关闭
`enable.replay`	是否开启数据回放功能	关闭
`session.timeout.ms`	消费者心跳丢失后的超时时间（`v3.3.3.0`起支持，范围`[6000, 1800000]`），超时触发 rebalance 后该 consumer 会被删除	`12000`
`max.poll.interval.ms`	消费者拉取数据间隔的最长时间（`v3.3.3.0`起支持，范围`[1000, INT32_MAX]`），超过则视为离线并触发 rebalance	`300000`
`fetch.max.wait.ms`	服务端单次返回数据的最大耗时（`v3.3.6.0`起支持，范围`[1, INT32_MAX]`）	`1000`
`min.poll.rows`	服务端单次返回数据的最小条数（`v3.3.6.0`起支持，范围`[1, INT32_MAX]`）	`4096`

高级参数（tmq_conf_new默认均为关闭）：

enable.wal.marker：提交位点时是否向 mnode 发送 WAL marker（boolean，默认false）。仅当提交的 offset 类型为 WAL log 时生效，一般用于迁移类工具。
msg.enable.batchmeta：是否启用服务端批量元数据返回（多条 meta 可合并为 batch meta 响应；非0开启，默认关闭）。Java WebSocket 侧属性名为enable_batch_meta。
msg.consume.rawdata：消费数据时拉取数据类型为二进制类型，不可做解析操作（内部参数，只用于 taosX 数据迁移；默认0不起效，非0起效，v3.3.6.0起支持）。

各语言连接器还提供各自专属参数，如 Go 的ws.url（支持多节点故障切换，如ws://node1:6041,ws://node2:6041）、timezone（IANA 时区，Go 连接器3.7.4及以上支持）等，完整列表见数据订阅编程接口。

订阅与消费流程

消费者订阅主题后，可以开始接收并处理这些主题中的消息。典型流程如下：

订阅数据：调用订阅接口，指定主题列表（名称），支持同时订阅多个主题。
拉取数据：调用 poll 类接口，每次调用获取一条消息；一条消息中可能包含多条记录。
解析结果：按各语言连接器约定解析消息字段；字段名和数据类型与主题定义的列一一对应。

限制提醒：订阅查询只能查询原始数据，不能查询聚合或计算结果；订阅查询只能按照时间正序查询数据。

指定订阅的 Offset 与提交 Offset

消费者可以指定从特定 Offset 开始读取分区中的消息，从而重读消息或跳过已处理的消息。当消费者读取并处理完消息后，可以提交 Offset，表示已经成功处理到该 Offset。Offset 提交可以是自动的（根据配置定期提交）或手动的（由应用程序控制何时提交）。

当创建消费者时属性enable.auto.commit为 false 时，可以手动提交 offset（如 C 语言的tmq_commit_sync、Rust 的consumer.commit）。
手工提交前确保消息正常处理完成，否则处理出错的消息不会被再次消费；自动提交是在本次poll消息时可能提交上次消息的消费进度，因此请确保消息处理完毕再进行下一次poll。

取消订阅和关闭消费

消费者可以取消对主题的订阅，停止接收消息。当消费者不再需要时，应关闭消费者实例，以释放资源并断开与 TDengine 服务器的连接。注意：消费者取消订阅后已经关闭，无法重用，如果想订阅新的 topic，请重新创建消费者。

各语言的完整可运行示例（WebSocket 与原生连接两种方式）见开发指南 · 数据订阅及仓库中的示例代码，如 c/tmq_demo.c、python/tmq_native.py、go/tmq/native/main.go 等。

回放功能（Replay）

TDengine 的数据订阅支持回放：按数据实际写入时间间隔重新推送消息，便于按原有节奏重放数据流。该能力基于 WAL 实现。

例如写入如下 3 条数据时，回放会先返回第 1 条，约 5s 后返回第 2 条，再约 3s 后返回第 3 条：

2023/09/22 00:00:00.000 2023/09/22 00:00:05.000 2023/09/22 00:00:08.000

使用回放功能时需要注意：

通过消费参数enable.replay为true开启回放。
仅查询主题支持回放，超级表主题和数据库主题不支持。
回放不支持进度保存。
回放本身需要处理时间，精度存在约数十毫秒的误差。
主题 SQL 中可用WHERE限定时间范围或过滤条件，这属于主题定义本身，与是否开启回放无关。

MQTT 订阅：通过 Bnode 使用标准 MQTT 客户端

TDengine 从v3.3.7.0开始提供 MQTT 订阅功能。通过 MQTT 客户端连接 TDengine Bnode 服务，可直接订阅系统中已有主题的数据。

主要特性：

协议支持：推荐使用 MQTT 5.0（亦兼容 MQTT 3.1 / 3.1.1；sub-offset等用户属性依赖 MQTT 5.0）。
身份验证：使用 TDengine 原生验证。
主题管理：与标准 MQTT 协议不同，主题必须预先创建（因不支持消息发布，无法通过发布消息动态创建）。
共享主题：形如$share/group_id/topic_name的主题被视为共享订阅，适用于需要负载均衡和高可用的场景。
订阅位置：支持latest、earliest（WAL 最早位置）；可通过订阅用户属性sub-offset=earliest指定，默认latest。
服务质量：支持 QoS 0、QoS 1。

Bnode 节点管理

用户可通过 TDengine 的命令行工具taos管理 Bnode。执行命令前，请确保taos可正常连接集群。

创建 Bnode：

CREATE BNODE ON DNODE {dnode_id}

一个 Dnode 上只能创建一个 Bnode。Bnode 创建成功后，会自动启动 Bnode 子进程taosmqtt，默认在 6057 端口对外提供 MQTT 订阅服务；端口可在taos.cfg中通过参数mqttPort配置（默认值 6057，取值范围 1～65056，局部配置参数，不支持动态修改），详见 taosd 配置参数 · mqttPort。该参数在 tglobal.c 中被注册为mqttPort配置项。例如：CREATE BNODE ON DNODE 1。

查看 Bnode：列出集群中所有的数据订阅节点，包括其id、endpoint、create_time等属性。更完整字段见元数据表INS_BNODES。

SHOW BNODES; taos> SHOW BNODES; id | endpoint | protocol | create_time | ==================================================================== 1 | 192.168.0.1:6057 | mqtt | 2024-11-28 18:44:27.089 | Query OK, 1 row(s) in set (0.037205s)

删除 Bnode：

DROP BNODE ON DNODE {dnode_id}

删除 Bnode 将把 Bnode 从 TDengine 集群中移除，同时停止taosmqtt服务。

订阅数据示例

环境准备：在taos中执行下面的 SQL，创建数据库、超级表、主题topic_meters、Bnode，并写入一条数据供下一步订阅使用。

CREATE DATABASE db VGROUPS 1; CREATE TABLE db.meters (ts TIMESTAMP, f1 INT) TAGS (t1 INT); CREATE TOPIC topic_meters AS SELECT ts, tbname, f1, t1 FROM db.meters; INSERT INTO db.tb USING db.meters TAGS (1) VALUES (now, 1); CREATE BNODE ON DNODE 1;

客户端订阅：可以使用兼容 MQTT 协议的客户端来订阅前一步环境中的数据，这里使用 Pythonpaho-mqtt举例说明（示例按 MQTT 5.0 编写，以便设置sub-offset）。在操作系统命令行中依次执行下面这些命令，即可订阅到上一步写入的数据；订阅成功后，若topic_meters主题中有新增写入，则会通过 MQTT 协议推送到客户端。

python3 -m venv .test-env source .test-env/bin/activate pip3 install paho-mqtt==2.1.0 python3 ./sub.py

其中sub.py文件的内容如下：

import time import paho.mqtt import paho.mqtt.properties as p import paho.mqtt.packettypes as pt import paho.mqtt.client as mqttClient def on_connect(client, userdata, flags, rc, properties=None): print("CONNACK received with code %s." % rc) sub_properties = p.Properties(pt.PacketTypes.SUBSCRIBE) sub_properties.UserProperty = ('sub-offset', 'earliest') client.subscribe("$share/g1/topic_meters", qos=1, properties=sub_properties) def on_subscribe(client, userdata, mid, granted_qos, properties=None): print("Subscribed: " + str(mid) + " " + str(granted_qos)) def on_message(client, userdata, msg): print(msg.topic + " " + str(msg.qos) + " " + str(msg.payload)) if paho.mqtt.__version__[0] > '1': client = mqttClient.Client(mqttClient.CallbackAPIVersion.VERSION2, client_id="tmq_sub_cid", userdata=None, protocol=mqttClient.MQTTv5) else: client = mqttClient.Client(client_id="tmq_sub_cid", userdata=None, protocol=mqttClient.MQTTv5) client.on_connect = on_connect client.username_pw_set("root", "taosdata") client.connect("127.0.1.1", 6057) client.on_subscribe = on_subscribe client.on_message = on_message client.loop_forever()

消息格式

上一节的示例中，会输出下面的信息：

CONNACK received with code Success. Subscribed: 1 [ReasonCode(Suback, 'Granted QoS 1')] topic_meters 1 b'{"topic":"topic_meters","db":"db","vid":2,"rows":[{"ts":1753086482326,"tbname":"tb","f1":1,"t1":1}]}'

其中第三行topic_meters是订阅的主题，1是该条消息的 QoS 值，后面是 UTF-8 编码的 JSON 消息，其中rows是数据行的数组。

深入原理与进一步阅读

从源码实现看，主题（topic）作为 mnode 管理的一等元数据对象被持久化在 SDB 中：mndTopic.c中定义了SDB_TOPIC存储类型及 topic 的序列化/反序列化逻辑，创建 topic 时即通过sdbGetSize校验实例级数量上限（mndTopic.c）。订阅数据的读取则发生在 vnode 侧：消费时依据消费进度从 WAL 直接读取，再交给统一查询引擎做过滤与变换，这与文档中"WAL 被改造成保留事件到达顺序的可持久化存储引擎"的描述相互印证。

如需继续深入，可以依次阅读：

主题语法：三种主题类型、WITH META/ONLY META语义与RELOAD TOPIC细节。
Native 订阅：消费者参数与消费流程的概念说明。
MQTT 订阅：Bnode 管理、MQTT 消息格式与完整示例。
数据订阅编程接口：各语言连接器（C、Java、Go、Rust、Python、C#、Node.js）的创建参数、WebSocket 与原生连接的完整代码示例。
元数据与性能表：INS_TOPICS、INS_SUBSCRIPTIONS、INS_BNODES、PERF_CONSUMERS，可用于查看主题、消费进度与消费者状态。

数据库
时序数据库
物联网
大数据
实时分析
云原生

【免费下载链接】tdengine

TDengine is an open source, high-performance, cloud native time-series database optimized for Internet of Things (IoT), Connected Cars, Industrial IoT and DevOps.

项目地址：https://gitcode.com/taosdata/tdengine

点击查看免费下载