首页
学习
活动
专区
工具
TVP
发布
精选内容/技术社群/优惠产品,尽在小程序
立即前往

关于 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 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 topics = new ArrayList();

topics.add("tmq_topic");

consumer.subscribe(topics);

消费

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

while(running){

ConsumerRecords 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 的开发者交流群(添加小T vx:tdengine),直接向社区技术支持人员寻求帮助。

  • 发表于:
  • 原文链接https://page.om.qq.com/page/OlHpHaUZ0Q4o9vl9ZZZx2D2w0
  • 腾讯「腾讯云开发者社区」是腾讯内容开放平台帐号(企鹅号)传播渠道之一,根据《腾讯内容开放平台服务协议》转载发布内容。
  • 如有侵权,请联系 cloudcommunity@tencent.com 删除。

扫码

添加站长 进交流群

领取专属 10元无门槛券

私享最新 技术干货

扫码加入开发者社群
领券