小T導讀：為了幫助應用實時獲取寫入時序數據庫（Time Series Database） TDengine 的(de)(de)數據(ju)(ju)，或(huo)者以事件到達順序(xu)處(chu)理數據(ju)(ju)，TDengine 提供了類似消息隊(dui)列(lie)產品的(de)(de)數據(ju)(ju)訂閱、消費接(jie)口。這樣在(zai)很多場景下，采用(yong) TDengine 的(de)(de)時(shi)序(xu)數據(ju)(ju)處(chu)理系(xi)(xi)統就不需要再集(ji)成(cheng)如(ru) Kafka 一般(ban)的(de)(de)消息隊(dui)列(lie)產品，從而簡化系(xi)(xi)統設計的(de)(de)復雜度，降低運維成(cheng)本(ben)。TDengine 3.0 對(dui)數據(ju)(ju)訂閱功能(neng)又進行了優化升(sheng)級，本(ben)文將詳細介紹其語法規則，方便(bian)開發者及企業使用(yong)。

與(yu) Kafka 一(yi)樣，應(ying)用 TDengine 時(shi)你也需要定義 topic, 但 TDengine 的(de) topic 是基于一(yi)個已經存在的(de)超級表、子表或普通表的(de)查(cha)詢條件(jian)，即一(yi)個?SELECT?語句。你可以(yi)(yi)使用 SQL 對(dui)標(biao)簽(qian)、表名(ming)、列、表達(da)式等條件(jian)進行過濾(lv)，以(yi)(yi)及對(dui)數(shu)(shu)據(ju)進行標(biao)量函數(shu)(shu)與(yu) UDF 計算（不包括數(shu)(shu)據(ju)聚(ju)合）。與(yu)其他消息(xi)隊列軟件(jian)相比，這是 TDengine 數(shu)(shu)據(ju)訂閱功(gong)能(neng)最大(da)的(de)優勢，它提(ti)供了更大(da)的(de)靈活性，數(shu)(shu)據(ju)的(de)顆粒度(du)可以(yi)(yi)由應(ying)用隨時(shi)調整(zheng)，而數(shu)(shu)據(ju)的(de)過濾(lv)與(yu)預(yu)處理(li)是交給(gei) TDengine 來(lai)完(wan)成，有效地減(jian)少傳輸(shu)的(de)數(shu)(shu)據(ju)量與(yu)應(ying)用的(de)復雜度(du)。

消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)訂閱 topic 后（一(yi)(yi)(yi)個(ge)(ge)(ge)消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)可(ke)以(yi)訂閱多(duo)個(ge)(ge)(ge) topic），可(ke)以(yi)實時獲得(de)最新的(de)數據。多(duo)個(ge)(ge)(ge)消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)可(ke)以(yi)組成一(yi)(yi)(yi)個(ge)(ge)(ge)消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)組 (consumer group)，一(yi)(yi)(yi)個(ge)(ge)(ge)消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)組里(li)的(de)多(duo)個(ge)(ge)(ge)消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)共享消(xiao)費(fei)(fei)(fei)(fei)(fei)進度，便于(yu)多(duo)線程(cheng)、分布式地消(xiao)費(fei)(fei)(fei)(fei)(fei)數據，提(ti)高消(xiao)費(fei)(fei)(fei)(fei)(fei)速(su)度；但(dan)不同消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)組中的(de)消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)即使消(xiao)費(fei)(fei)(fei)(fei)(fei)同一(yi)(yi)(yi)個(ge)(ge)(ge) topic，也并(bing)不共享消(xiao)費(fei)(fei)(fei)(fei)(fei)進度。如果訂閱的(de)是超級表，數據可(ke)能(neng)(neng)會分布在多(duo)個(ge)(ge)(ge)不同的(de) vnode 上(shang)，也就是多(duo)個(ge)(ge)(ge) shard 上(shang)，這樣一(yi)(yi)(yi)個(ge)(ge)(ge)消(xiao)費(fei)(fei)(fei)(fei)(fei)組里(li)有多(duo)個(ge)(ge)(ge)消(xiao)費(fei)(fei)(fei)(fei)(fei)者(zhe)(zhe)(zhe)(zhe)可(ke)以(yi)提(ti)高消(xiao)費(fei)(fei)(fei)(fei)(fei)效率。TDengine 的(de)消(xiao)息隊列提(ti)供了(le)消(xiao)息的(de) ACK 機制，在宕機、重啟等復雜環境下(xia)也能(neng)(neng)確(que)保 at least once 消(xiao)費(fei)(fei)(fei)(fei)(fei)。

為了實現上述功能，TDengine 會為 WAL (Write-Ahead-Log) 文(wen)件(jian)自(zi)動創建索引以(yi)支持快速(su)隨機訪(fang)問，并提供(gong)了靈活可(ke)配(pei)置的(de)文(wen)件(jian)切換與保留機制，用戶可(ke)以(yi)按需指定 WAL 文(wen)件(jian)保留的(de)時間(jian)以(yi)及大小(xiao)：

• WAL_RETENTION_PERIOD：為了數據訂閱消費，需要 WAL 日志文件額外保留的最大時長策略。WAL 日志清理，不受訂閱客戶端消費狀態影響。單位為 s，默認為 3600，表示在 WAL 保留最近 3600 秒的數據，用戶可以根據數據訂閱的需要修改這個參數為適當值。
• WAL_RETENTION_SIZE：為了數據訂閱消費，需要 WAL 日志文件額外保留的最大累計大小策略。單位為 KB，默認為 0，表示累計大小無上限。

通過以上方式，我們將 WAL 改造成了一個保留事件到達順序的、可持久化的存儲引擎（但由于 TSDB 具有遠比 WAL 更高的壓縮率，因此不推薦保留太長時間，一般來說建議不超過幾天）。對于以 topic 形式創建的查詢，TDengine 將對接 WAL 而不是 TSDB 作為其存儲引擎。在消費時，TDengine 根據當前消費進度從 WAL 直接讀取數據，并使用統一的查詢引擎實現過濾、變換等操作，將數據推送給消費者。
為(wei)了方(fang)便大(da)家(jia)上(shang)手(shou)實操，下文將對(dui) TDengine 數(shu)據訂閱相關語法進行詳細解讀。

寫入數據

首先完成(cheng)建庫、建一張(zhang)超級表(biao)和多張(zhang)子表(biao)操作(zuo)，然后就可以寫入數據了，比如：

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 創建如下所(suo)示 topic（topic 創建個數有(you)上限，通過參數 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 語句訂閱(yue)（包括 SELECT *，或(huo) SELECT ts, c1 等指定(ding)列訂閱(yue)，可以帶條件過濾、標(biao)量函數計算，但不支(zhi)持(chi)聚合(he)函數、不支(zhi)持(chi)時間(jian)窗(chuang)口聚合(he)）。但需要(yao)注意的是：

• 該類型 TOPIC 一旦創建則訂閱數據的結構確定；
• 被訂閱或用于計算的列或標簽不可被刪除（ALTER table DROP）、修改（ALTER table MODIFY）；
• 若發生表結構變更，新增的列不出現在結果中。

超級表訂閱

CREATE TOPIC topic_name AS STABLE stb_name

與(yu) SELECT * from stbName 訂閱的(de)區別是：

• 不會限制用戶的表結構變更。
• 返回的是非結構化的數據：返回數據的結構會隨超級表的表結構變化而變化。
• 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;

通過該語句可創建一個包含數(shu)據庫所有(you)表數(shu)據的訂(ding)閱，with meta 參數(shu)可選，同上。

創建消費者

訂閱?topics

一個 consumer 支持(chi)同時訂(ding)閱多個 topic。以 Java 為(wei)例：

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

消費

在 Java 語(yu)言下如何對 TMQ 消息(xi)進(jin)行消費，代碼示(shi)意如下：

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

結束消費

消費結(jie)束(shu)后，應當取消訂閱。

/* 取消訂閱 */
tmq_unsubscribe(tmq);

/* 關閉消費者對象 */
tmq_consumer_close(tmq);

刪除 topic

如果(guo)不再需要(yao)訂(ding)閱數據，可以刪除(chu) topic，需要(yao)注意：只有當前未在訂(ding)閱中的 topic 才能(neng)被(bei)刪除(chu)。

/* 刪除 topic */
DROP TOPIC topic_name;

狀態查看

1、topics：查詢已經(jing)創建的 topic

SHOW TOPICS;

2、consumers：查詢(xun) consumer 的狀態及其訂閱的 topic

SHOW CONSUMERS;

3、subscriptions：查詢(xun) consumer 與 vgroup 之間的分(fen)配關(guan)系

SHOW SUBSCRIPTIONS;

寫在最后

受文章(zhang)篇幅所限(xian)，本(ben)文只(zhi)分(fen)(fen)享了部分(fen)(fen)語法的(de)(de)具體實現，需要了解(jie)相關(guan)設置及(ji)更多語言的(de)(de)代(dai)碼示例，可以進入(ru) TDengine 官網查詢數(shu)據訂閱的(de)(de)。對于更為(wei)復(fu)雜的(de)(de)應用問(wen)題，也歡迎大家加入(ru) TDengine 的(de)(de)開發(fa)者交流群（添(tian)加小T vx：tdengine），直接向(xiang)社區(qu)技(ji)術(shu)支(zhi)持人員尋求(qiu)幫助。