眾所周(zhou)知，在 TDengine 3.0 中，我們對數(shu)(shu)據(ju)(ju)訂(ding)閱(yue)功(gong)能(neng)進行了(le)全面升級，以便大家可以更(geng)加便捷地實(shi)時訂(ding)閱(yue)和(he)獲取數(shu)(shu)據(ju)(ju)的更(geng)新，完成實(shi)時監控(kong)、數(shu)(shu)據(ju)(ju)分析(xi)和(he)有效報警等(deng)工(gong)作(zuo)。在本文中，TDengine 資深研發將(jiang)以 TDengine 3.0 為(wei)對象，為(wei)大家介紹數(shu)(shu)據(ju)(ju)訂(ding)閱(yue)功(gong)能(neng)的正確打開方式，給到有需要的人作(zuo)參考指南，避(bi)免走(zou)入(ru)應用誤區。

本文將從 Java Developer 的視角來介紹如何(he)使用 TDengine 3.0 的數(shu)據訂閱功能。

TDengine 3.0 的版本迭代很快，可能有些配置參數或細節在之后的版本會發生變化，本文對應 TDengine 版本為 3.0.3.0。

寫在前面

在官方文檔里已經有介(jie)紹，TDengine 的(de)(de)數據訂閱是什么以(yi)(yi)及如何使用，有需(xu)要的(de)(de)朋友可(ke)以(yi)(yi)通(tong)過下方鏈接進入官網查(cha)看相關介(jie)紹：

1. 開發指南：
2. SQL：
3. Java 使用數據訂閱：

總結一下(xia)，我理解的(de)(de)(de)數據(ju)訂(ding)(ding)(ding)閱功能是以(yi)“訂(ding)(ding)(ding)閱”的(de)(de)(de)方式獲取(qu)存在(zai)于 TDengine 中的(de)(de)(de)數據(ju)。一般情(qing)況下(xia)，“訂(ding)(ding)(ding)閱”意味著(zhu)的(de)(de)(de)業務需(xu)求是訂(ding)(ding)(ding)閱數據(ju)庫中的(de)(de)(de)最新數據(ju)。“訂(ding)(ding)(ding)閱”的(de)(de)(de)流程很簡單：（1）在(zai)數據(ju)庫中創建 topic；（2）在(zai)應用中消費 topic 的(de)(de)(de)數據(ju)。

基本操作：創建

在數據庫中創建 topic，使用 SQL 語句 create topic 即可。create topic 這個 SQL 如何寫，實際上定義了 topic 對應的數據粒度，包括哪些數據庫、超級表、子表、列、行。值得一提的是，TDengine 的 SQL 支持訂閱 database、 supertable、subquery 這 3 種模式。CREATE TOPIC topic_name [WITH META] AS DATABASE db_name; 這種 SQL 可以直接訂閱整個 database；CREATE TOPIC topic_name AS STABLE stb_name 這種 SQL 可以訂閱某個(ge)超級(ji)表；訂閱子查詢是最普遍的場(chang)景。例如(ru)：

CREATE TOPIC topic_name AS SELECT ts,voltage,location FROM testdb.meters WHERE voltage > 220.0 and location in ('北京','天津');

上(shang)面這個 SQL，訂閱了 testdb 數(shu)據庫(ku)中的 meters 超級表(biao)，通(tong)過(guo) where 子句過(guo)濾滿足以下條件：location（tag 列）為“北京(jing)”或“天津”的子表(biao)，且 voltage 超過(guo) 220.0 的 ts、voltage、location 的數(shu)據。

黃金搭檔：流式計算 + 數據訂閱

以智能電表(biao)的(de)(de)場(chang)景為(wei)例，如果我(wo)想(xiang)每 10 分鐘計算(suan)一(yi)次電壓(ya)的(de)(de)平均值，并(bing)在(zai)平均電壓(ya)高于 220V 就進行上報。對于這種需(xu)求，單純用(yong)(yong) TDengine 的(de)(de)數據(ju)(ju)訂閱(yue)功能是不(bu)行的(de)(de)，因為(wei) create topic 的(de)(de)子查詢(xun)(xun)不(bu)支持(chi)聚(ju)合查詢(xun)(xun)。這個時候，就需(xu)要(yao)用(yong)(yong) TDengine 的(de)(de)流式(shi)計算(suan) + 數據(ju)(ju)訂閱(yue)這對黃金(jin)搭檔了。如下：

CREATE STREAM stream_name TRIGGER WINDOW_CLOSE IGNORE EXPIRED 1 
INTO stb_name 
AS SELECT _wend as ts, avg(voltage) as voltage, last_row(location) as location
FROM testdb.meters 
WHERE location in ('北京', '天津') 
PARTITION BY location 
INTERVAL(10m);

CREATE TOPIC topic_name AS SELECT * FROM stream_name where voltage > 220.0;

上面的(de) 2 條 SQL 中(zhong)，第(di)一(yi)條 SQL 創建(jian)了一(yi)個 stream：以 location 分組，計算(suan)每(mei) 10 分鐘的(de)“北京(jing)”、“天津”的(de)平均電壓；用時(shi)(shi)間窗口的(de)結束 _wend 作為時(shi)(shi)間戳 ts；avg(voltage) 計算(suan) voltage 平均值；時(shi)(shi)間窗口的(de)最后一(yi)條 last_row(location) 作為標簽。同時(shi)(shi)，這個 stream 以 WINDOW_CLOSE 作為計算(suan)窗口的(de)觸發(fa)模式，過期策(ce)略為 IGNORE EXPIRED 1。

第二條 SQL 創(chuang)建了(le)子查詢訂閱，用于過濾每 10 分(fen)鐘平均(jun)電(dian)壓高(gao)于 220V 的數據。這樣我(wo)們就創(chuang)建了(le)一個可以被消(xiao)費(fei)的 topic，消(xiao)費(fei)到(dao)的數據為高(gao)于 220V 的 10 分(fen)鐘平均(jun)電(dian)壓，滿(man)足了(le)前面所(suo)說(shuo)的監控場景的需(xu)求。

消費 topic：很像 Kafka

在(zai)應用(yong)中消(xiao)費(fei) topic 的數(shu)據，需要按照各種連接器的 API 來使(shi)用(yong)，具體使(shi)用(yong)方式請參(can)考官方文檔：。在(zai)這里(li)，我只對 TDengine 和訂閱消(xiao)費(fei) topic 的一些配置參(can)數(shu)進(jin)行梳理。

1. 連接相關的參數，java connector 中使用 bootstrap.servers 一個參數代替了 td.connect.ip 和 td.connect.port，使用了和 Kafka 一樣的參數名。td.connect.user 和 td.connect.pass 仍然需要設置。
2. group.id：和 Kafka 一樣，多個線程可以共同消費同一個 topic，只要它們使用同一個 group.id。TDengine 的 vgroup 與 Kafka 的 partition 在概念上是對應的。同一個 group.id 中，一個 vgroup 最多只對應一個 consumer。如果 consumer 數量大于 vgroup 的數量，則有些 consumer 消費不到數據。
3. auto.offset.reset：這個參數和 Kafka 的行為不一樣。如果 group.id 為新值，在設置 earliest 時，訂閱從頭消費數據；設置為 latest 時，從最新數據開始訂閱。當 group.id 為已存在的值時，不管 auto.offset.reset 為何值，都會從最后一個 offset 開始，繼續消費。
4. enable.auto.commit：建議設置為 false。開啟自動提交 offset，TDengine 的 commit 自動提交機制是輪詢提交。
5. auto.commit.interval.ms：建議不設置。如果 enable.auto.commit 為 true，自動提交 commit 的間隔為 auto.commit.interval.ms 設置的值。
6. enable.heartbeat.background：建議設置為 true，默認值為 true。如果設置為 false，在應用長時間不主動 poll 數據時，可能會造成當前 consumer 的離線。在 TDengine 的實現上，heartbeat 的 interval 被設置成了 1 秒。
7. msg.with.table.name：建議設置成 true。在訂閱超級表和數據庫時添加了 WITH META，應該開啟這個設置。例如：訂閱為 CREATE TOPIC topic_name WITH META AS STABLE stb 時，配置 msg.with.table.name 為 true，則消費時可以獲取到 tableName。

Show U The Code

到(dao)此，本文(wen)介紹了有關(guan) TDengine3.0 的數據(ju)訂(ding)閱功能(neng)的諸(zhu)多細節。我相信(xin)，上(shang)面的內容應該可(ke)以為(wei)你使用(yong)數據(ju)訂(ding)閱功能(neng)提供(gong)一(yi)(yi)些思路和幫助。但是，對(dui)程序員(yuan)來說，“Talk is cheap. Show me the code”。下面，我列舉了一(yi)(yi)些 Java 的示(shi)例代碼，供(gong)你參(can)考。

1. subscribeDemo-java

這個 java 工程實現了一個最(zui)簡單的(de)訂閱(yue)功能，從 TDengine 中(zhong)訂閱(yue)一個 topic ，并將消費(fei)到的(de)數據(ju)寫(xie)(xie)到文件中(zhong)。值得一提的(de)是，代碼(ma)使用 動態生成了 Java POJO 類和對(dui)應的(de) Deserializer 類。因此，你只需要在(zai) schema.txt 內寫(xie)(xie)好 topic 對(dui)應的(de)字段，就可以(yi)不(bu)寫(xie)(xie)代碼(ma)，直接(jie)訂閱(yue)不(bu)同 topic 的(de)數據(ju)了。

鏈接：

2. SubscribeDemo

這個頁面展示了一段最基本的(de)數(shu)據(ju)訂閱的(de)代碼。main 方法(fa)中，包括了在 TDengine 中創建(jian)數(shu)據(ju)庫(ku)、表、topic 的(de)操作，并從 topic 中消(xiao)費數(shu)據(ju)進行打印。

鏈接：

3. WebsocketSubscribeDemo

這個(ge)頁面的代碼和 SubscribeDemo 相比，僅(jin)有的區別是其配(pei)置(zhi)了 td.connect.type 參數為 ws，即：使用 websocket 連(lian)接 taosadapter，這樣的好處是不用安裝客(ke)戶(hu)端。

鏈接：

結語

相信借助本篇文章，你一定能夠流暢體驗到 TDengine 的數據訂閱功能，有需要的讀者可以收藏備用。對于更為復雜的應用問題，也歡迎大家加入 TDengine 的開發者交流群（添加小T vx：tdengine），直接向社區(qu)技術(shu)支持人員尋求幫助。關(guan)于 TDengine 3.0 的更多示例代碼(ma)，請參(can)考：。