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

<!--StartFragment--> > 小T导读：为了帮助应用实时获取写入[时序数据库](https://www.taosdata.com/ "时序数据库")（[Time Series Database](https://www.taosdata.com/time-series-database "Time Series DataBase")） [TDengine](https://www.taosdata.com/ "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](https://www.taosdata.com/tsdb "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 WHEREc1 > 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 官网查询数据订阅的[相关文档](https://docs.taosdata.com/develop/tmq/)。对于更为复杂的应用问题，也欢迎大家加入 TDengine 的开发者交流群，直接向社区技术支持人员寻求帮助。 <!--EndFragment-->