小T导读
WAL_RETENTION_PERIOD：为了数据订阅消费，需要 WAL 日志文件额外保留的最大时长策略。WAL 日志清理，不受订阅客户端消费状态影响。单位为 s，默认为 3600，表示在 WAL 保留最近 3600 秒的数据，用户可以根据数据订阅的需要修改这个参数为适当值。
WAL_RETENTION_SIZE：为了数据订阅消费，需要 WAL 日志文件额外保留的最大累计大小策略。单位为 KB，默认为 0，表示累计大小无上限。
通过以上方式，我们将 WAL 改造成了一个保留事件到达顺序的、可持久化的存储引擎（但由于 TSDB 具有远比 WAL 更高的压缩率，因此不推荐保留太长时间，一般来说建议不超过几天）。对于以 topic 形式创建的查询，TDengine 将对接 WAL 而不是 TSDB 作为其存储引擎。在消费时，TDengine 根据当前消费进度从 WAL 直接读取数据，并使用统一的查询引擎实现过滤、变换等操作，将数据推送给消费者。
01
写入数据
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');
02
创建 topic
CREATE TOPIC topic_name AS SELECT ts, c1, c2, c3 FROM tmqdb.stb WHEREc1 > 1;TMQ 支持以下多种订阅类型：
列订阅
CREATE TOPIC topic_name as subquery该类型 TOPIC 一旦创建则订阅数据的结构确定；
被订阅或用于计算的列或标签不可被删除（ALTER table DROP）、修改（ALTER table MODIFY）；
若发生表结构变更，新增的列不出现在结果中。
超级表订阅
CREATE TOPIC topic_name AS STABLE stb_name不会限制用户的表结构变更。
返回的是非结构化的数据：返回数据的结构会随超级表的表结构变化而变化。
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;03
创建消费者
订阅 topics
List<String> topics = new ArrayList<>();topics.add("tmq_topic");consumer.subscribe(topics);
消费
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 */DROP TOPIC topic_name;
状态查看
SHOW TOPICS;SHOW CONSUMERS;SHOW SUBSCRIPTIONS;04
写在最后
往
期
推
荐
本文分享自微信公众号 - TDengine（taosdata_news）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
