关于 TDengine 3.0 数据订阅，你需要知道这些

小 T 导读：为了帮助应用实时获取写入时序数据库（Time Series Database） TDengine 的数据，或者以事件到达顺序处理数据，TDengine 提供了类似消息队列产品的数据订阅、消费接口。这样在很多场景下，采用 TDengine 的时序数据处理系统就不需要再集成如 Kafka 一般的消息队列产品，从而简化系统设计的复杂度，降低运维成本。TDengine 3.0 对数据订阅功能又进行了优化升级，本文将详细介绍其语法规则，方便开发者及企业使用。

与 Kafka 一样，应用 TDengine 时你也需要定义 topic, 但 TDengine 的 topic 是基于一个已经存在的超级表、子表或普通表的查询条件，即一个 SELECT 语句。你可以使用 SQL 对标签、表名、列、表达式等条件进行过滤，以及对数据进行标量函数与 UDF 计算（不包括数据聚合）。与其他消息队列软件相比，这是 TDengine 数据订阅功能最大的优势，它提供了更大的灵活性，数据的颗粒度可以由应用随时调整，而数据的过滤与预处理是交给 TDengine 来完成，有效地减少传输的数据量与应用的复杂度。

消费者订阅 topic 后（一个消费者可以订阅多个 topic），可以实时获得最新的数据。多个消费者可以组成一个消费者组 (consumer group)，一个消费者组里的多个消费者共享消费进度，便于多线程、分布式地消费数据，提高消费速度；但不同消费者组中的消费者即使消费同一个 topic，也并不共享消费进度。如果订阅的是超级表，数据可能会分布在多个不同的 vnode 上，也就是多个 shard 上，这样一个消费组里有多个消费者可以提高消费效率。TDengine 的消息队列提供了消息的 ACK 机制，在宕机、重启等复杂环境下也能确保 at least once 消费。

为了实现上述功能，TDengine 会为 WAL (Write-Ahead-Log) 文件自动创建索引以支持快速随机访问，并提供了灵活可配置的文件切换与保留机制，用户可以按需指定 WAL 文件保留的时间以及大小：

• WAL_RETENTION_PERIOD：为了数据订阅消费，需要 WAL 日志文件额外保留的最大时长策略。WAL 日志清理，不受订阅客户端消费状态影响。单位为 s，默认为 3600，表示在 WAL 保留最近 3600 秒的数据，用户可以根据数据订阅的需要修改这个参数为适当值。

• WAL_RETENTION_SIZE：为了数据订阅消费，需要 WAL 日志文件额外保留的最大累计大小策略。单位为 KB，默认为 0，表示累计大小无上限。

通过以上方式，我们将 WAL 改造成了一个保留事件到达顺序的、可持久化的存储引擎（但由于 TSDB 具有远比 WAL 更高的压缩率，因此不推荐保留太长时间，一般来说建议不超过几天）。对于以 topic 形式创建的查询，TDengine 将对接 WAL 而不是 TSDB 作为其存储引擎。在消费时，TDengine 根据当前消费进度从 WAL 直接读取数据，并使用统一的查询引擎实现过滤、变换等操作，将数据推送给消费者。

为了方便大家上手实操，下文将对 TDengine 数据订阅相关语法进行详细解读。

写入数据

首先完成建库、建一张超级表和多张子表操作，然后就可以写入数据了，比如：

DROP DATABASE IF EXISTS tmqdb;CREATE DATABASE tmqdb;CREATE TABLE tmqdb.stb (ts TIMESTAMP, c1 INT, c2 FLOAT, c3 VARCHAR(16) TAGS(t1 INT, t3 VARCHAR(16));CREATE TABLE tmqdb.ctb0 USING tmqdb.stb TAGS(0, "subtable0");CREATE TABLE tmqdb.ctb1 USING tmqdb.stb TAGS(1, "subtable1");       INSERT INTO tmqdb.ctb0 VALUES(now, 0, 0, 'a0')(now+1s, 0, 0, 'a00');INSERT INTO tmqdb.ctb1 VALUES(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11');

创建 topic

TDengine 使用 SQL 创建如下所示 topic（topic 创建个数有上限，通过参数 tmqMaxTopicNum 控制，默认 20 个）：

CREATE TOPIC topic_name AS SELECT ts, c1, c2, c3 FROM tmqdb.stb WHERE c1 > 1;

TMQ 支持以下多种订阅类型：

列订阅

CREATE TOPIC topic_name as subquery

通过 SELECT 语句订阅（包括 SELECT *，或 SELECT ts, c1 等指定列订阅，可以带条件过滤、标量函数计算，但不支持聚合函数、不支持时间窗口聚合）。但需要注意的是：

• 该类型 TOPIC 一旦创建则订阅数据的结构确定；

• 被订阅或用于计算的列或标签不可被删除（ALTER table DROP）、修改（ALTER table MODIFY）；

• 若发生表结构变更，新增的列不出现在结果中。

超级表订阅

CREATE TOPIC topic_name AS STABLE stb_name

与 SELECT * from stbName 订阅的区别是：

• 不会限制用户的表结构变更。

• 返回的是非结构化的数据：返回数据的结构会随超级表的表结构变化而变化。

• with meta 参数可选，选择时将返回创建超级表，子表等语句，主要用于 taosx 做超级表迁移。

• where_condition 参数可选，选择时将用来过滤符合条件的子表，订阅这些子表。where 条件里不能有普通列，只能是 tag 或 tbname，where 条件里可以用函数，用来过滤 tag，但是不能是聚合函数，因为子表 tag 值无法做聚合。也可以是常量表达式，比如 2 > 1（订阅全部子表），或者 false（订阅 0 个子表）。

• 返回数据不包含标签。

数据库订阅

CREATE TOPIC topic_name [WITH META] AS DATABASE db_name;

通过该语句可创建一个包含数据库所有表数据的订阅，with meta 参数可选，同上。

创建消费者

订阅 topics

一个 consumer 支持同时订阅多个 topic。以 Java 为例：

List<String> topics = new ArrayList<>();topics.add("tmq_topic");consumer.subscribe(topics);

消费

在 Java 语言下如何对 TMQ 消息进行消费，代码示意如下：

while(running){  ConsumerRecords<Meters> meters = consumer.poll(Duration.ofMillis(100));    for (Meters meter : meters) {      processMsg(meter);    }    }

结束消费

消费结束后，应当取消订阅。

/* 取消订阅 */tmq_unsubscribe(tmq);
/* 关闭消费者对象 */tmq_consumer_close(tmq);

删除 topic

如果不再需要订阅数据，可以删除 topic，需要注意：只有当前未在订阅中的 topic 才能被删除。

/* 删除 topic */DROP TOPIC topic_name;

状态查看

1、topics：查询已经创建的 topic

SHOW TOPICS;

2、consumers：查询 consumer 的状态及其订阅的 topic

SHOW CONSUMERS;

3、subscriptions：查询 consumer 与 vgroup 之间的分配关系

SHOW SUBSCRIPTIONS;

写在最后

受文章篇幅所限，本文只分享了部分语法的具体实现，需要了解相关设置及更多语言的代码示例，可以进入 TDengine 官网查询数据订阅的相关文档。对于更为复杂的应用问题，也欢迎大家加入 TDengine 的开发者交流群，直接向社区技术支持人员寻求帮助。