如何使用規(guī)則引擎存儲消息到OpenTSDB數(shù)據(jù)庫，很多新手對此不是很清楚，為了幫助大家解決這個(gè)難題，下面小編將為大家詳細(xì)講解，有這方面需求的人可以來學(xué)習(xí)下，希望你能有所收獲。

桐梓網(wǎng)站建設(shè)公司創(chuàng)新互聯(lián),桐梓網(wǎng)站設(shè)計(jì)制作，有大型網(wǎng)站制作公司豐富經(jīng)驗(yàn)。已為桐梓超過千家提供企業(yè)網(wǎng)站建設(shè)服務(wù)。企業(yè)網(wǎng)站搭建\外貿(mào)網(wǎng)站制作要多少錢，請找那個(gè)售后服務(wù)好的桐梓做網(wǎng)站的公司定做！

OpenTSDB 介紹

OpenTSDB 是可擴(kuò)展的分布式時(shí)序數(shù)據(jù)庫，底層依賴 HBase 并充分發(fā)揮了HBase的分布式列存儲特性，支持?jǐn)?shù)百萬每秒的讀寫。

面對大規(guī)?？焖僭鲩L的物聯(lián)網(wǎng)傳感器采集、交易記錄等數(shù)據(jù)，時(shí)間序列數(shù)據(jù)累計(jì)速度非常快，時(shí)序數(shù)據(jù)庫通過提高效率來處理這種大規(guī)模數(shù)據(jù)，并帶來性能的提升，包括：更高的容納率（Ingest Rates）、更快的大規(guī)模查詢以及更好的數(shù)據(jù)壓縮。

安裝與驗(yàn)證 OpenTSDB 服務(wù)器

讀者可以參考 OpenTSDB 官方文檔 (http://opentsdb.net) 或 Docker (https://hub.docker.com/r/petergrace/opentsdb-docker/) 來下載安裝 OpenTSDB 服務(wù)器，本文使用 OpenTSDB 2.4.0 版本。

場景介紹

該場景需要將 EMQ X 指定主題下且滿足條件的消息存儲到 OpenTSDB 數(shù)據(jù)庫。為了便于后續(xù)分析檢索，消息內(nèi)容需要進(jìn)行拆分存儲。

該場景下客戶端上報(bào)數(shù)據(jù)如下：

• Topic：stat/cpu

• Payload:

  {
    "metric": "cpu",
    "tags": {
      "host": "serverA"
    },
    "value":12
  }

準(zhǔn)備工作

啟動 OpenTSDB Server

啟動 OpenTSDB Server 并開放 4242 端口。

$ docker pull petergrace/opentsdb-docker

$ docker run -d --name opentsdb -p 4242:4242 petergrace/opentsdb-docker

配置說明

創(chuàng)建資源

打開 EMQ X Dashboard，進(jìn)入左側(cè)菜單的 資源頁面，點(diǎn)擊 新建按鈕，選擇 OpenTSDB 資源類型并完成相關(guān)配置進(jìn)行資源創(chuàng)建。

創(chuàng)建規(guī)則

進(jìn)入左側(cè)菜單的 規(guī)則頁面，點(diǎn)擊 新建按鈕，進(jìn)行規(guī)則創(chuàng)建。這里選擇觸發(fā)事件 message.publish，即在 EMQ X 收到 PUBLISH 消息時(shí)觸發(fā)該規(guī)則進(jìn)行數(shù)據(jù)處理。

選定觸發(fā)事件后，我們可在界面上看到可選字段及示例 SQL：

篩選所需字段

規(guī)則引擎使用 SQL 語句過濾和處理數(shù)據(jù)。例如前文提到的場景中我們需要將 payload 中的字段提取出來使用，則可以通過 payload. 實(shí)現(xiàn)。同時(shí)我們僅僅期望處理 stat/cpu 主題，那么可以在 WHERE 子句中使用主題通配符 =~ 對 topic 進(jìn)行篩選：topic =~ 'stat/cpu'， 最終我們得到 SQL 如下：

SELECT
  payload.metric as metric, payload.tags as tags, payload.value as value
FROM
  "message.publish"
WHERE
	topic =~ 'stat/cpu'

SQL 測試

借助 SQL 測試功能，我們可以快速確認(rèn)剛剛填寫的 SQL 語句能否達(dá)成我們的目的。首先填寫用于測試的 payload 等數(shù)據(jù)如下：

然后點(diǎn)擊 測試按鈕，我們得到以下數(shù)據(jù)輸出：

{
  "metric": "cpu",
  "tags": {
    "host": "serverA"
  },
  "value": 12
}

測試輸出與預(yù)期相符，我們可以進(jìn)行后續(xù)步驟。

添加響應(yīng)動作，存儲消息到 OpenTSDB

SQL 條件輸入輸出無誤后，我們繼續(xù)添加相應(yīng)動作，配置寫入 SQL 語句，將篩選結(jié)果存儲到 OpenTSDB。

點(diǎn)擊響應(yīng)動作中的 添加按鈕，選擇 保存數(shù)據(jù)到 OpenTSDB動作，選取剛剛創(chuàng)建的 OpenTSDB 資源并完成剩余參數(shù)設(shè)置。OpenTSDB 動作用到的幾個(gè)參數(shù)分別為：

1. 詳細(xì)信息。是否需要 OpenTSDB Server 返回存儲失敗的 Data points 及失敗原因，默認(rèn)為 false。

2. 摘要信息。是否需要 OpenTSDB Server 返回 data point 存儲成功與失敗的數(shù)量，默認(rèn)為 true。

3. 最大批處理數(shù)量。消息請求頻繁時(shí)允許驅(qū)動從隊(duì)列中一次讀取多少個(gè) Data Points 合并為一個(gè) HTTP 請求，為性能優(yōu)化參數(shù)，默認(rèn)為 20。

4. 是否同步調(diào)用。配置 OpenTSDB Server 是否等待所有數(shù)據(jù)都被寫入后才返回結(jié)果，默認(rèn)為 false。

5. 同步調(diào)用超時(shí)時(shí)間。OpenTSDB Server 等待數(shù)據(jù)寫入的最大時(shí)間，默認(rèn)為 0，即永不超時(shí)。

這里我們?nèi)渴褂媚J(rèn)配置，點(diǎn)擊 新建按鈕完成規(guī)則創(chuàng)建。

測試

預(yù)期結(jié)果

我們成功創(chuàng)建了一條規(guī)則，包含一個(gè)處理動作，動作期望效果如下：

1. 客戶端向 stat/cpu 主題上報(bào)消息時(shí)，該消息將命中 SQL，規(guī)則列表中 已命中數(shù)字增加 1；

2. OpenTSDB Server 中將增加一條數(shù)據(jù)，數(shù)據(jù)內(nèi)容與消息內(nèi)容一致。

使用 Dashboard 中的 Websocket 工具測試

切換到 工具--> Websocket頁面，使用任意信息客戶端連接到 EMQ X，連接成功后在 消息卡片中發(fā)送如下消息：

• Topic：stat/cpu

• Payload:

  {
    "metric": "cpu",
    "tags": {
      "host": "serverA"
    },
    "value":12
  }

點(diǎn)擊 發(fā)送按鈕，發(fā)送成功后可以看到當(dāng)前規(guī)則已命中次數(shù)已經(jīng)變?yōu)榱?1。

然后通過 Postman 向 OpenTSDB 發(fā)送查詢請求，當(dāng)我們得到如下應(yīng)答時(shí)說明新的 data point 已經(jīng)添加成功：

至此，我們通過規(guī)則引擎實(shí)現(xiàn)了使用規(guī)則引擎存儲消息到 OpenTSDB 數(shù)據(jù)庫的業(yè)務(wù)開發(fā)。

看完上述內(nèi)容是否對您有幫助呢？如果還想對相關(guān)知識有進(jìn)一步的了解或閱讀更多相關(guān)文章，請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝您對創(chuàng)新互聯(lián)的支持。