如何通過(guò)InfluxDB來(lái)存儲(chǔ)相關(guān)的信息

本篇內(nèi)容主要講解“如何通過(guò)InfluxDB來(lái)存儲(chǔ)相關(guān)的信息”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“如何通過(guò)InfluxDB來(lái)存儲(chǔ)相關(guān)的信息”吧!

InfluxDB 是一個(gè)由 InfluxData 開(kāi)發(fā)的開(kāi)源時(shí)序型數(shù)據(jù)庫(kù)。 它由 Go 寫成，著力于高性能地查詢與存儲(chǔ)時(shí)序型數(shù)據(jù)，相比上一期中介紹的 OpenTSDB 數(shù)據(jù)庫(kù) InfluxDB 較為輕量，在 InfluxData 官方給出的各項(xiàng)指標(biāo)基準(zhǔn)測(cè)試用 InfluxDB 都強(qiáng)于 OpenTSDB。

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

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

下載安裝 InfluxDB 服務(wù)器，本文使用 InfluxDB 1.7 版本。

配置 EMQ X 服務(wù)器

通過(guò) RPM 方式安裝的 EMQ X，InfluxDB 相關(guān)的配置文件位于 /etc/emqx/plugins/emqx_backend_influxdb.conf，考慮到功能定位，InfluxDB 插件僅支持消息存儲(chǔ)功能。

配置連接地址與連接池大?。?/strong>

## InfluxDB UDP Server
## 僅使用 UDP 接入
backend.influxdb.pool1.server = 127.0.0.1:8089

## InfluxDB Pool Size
backend.influxdb.pool1.pool_size = 5

## Whether or not set timestamp when encoding InfluxDB line
backend.influxdb.pool1.set_timestamp = trues

**InfluxDB Backend 消息存儲(chǔ)規(guī)則參數(shù): **

通過(guò) topic 過(guò)濾器，設(shè)置需要存儲(chǔ)消息的主題，pool 參數(shù)區(qū)別多個(gè)數(shù)據(jù)源：

## Store Publish Message
backend.influxdb.hook.message.publish.1 = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}

啟動(dòng)該插件，啟動(dòng)插件的方式有 命令行和 控制臺(tái)兩種方式，用戶可以任選其一。

消息模板

由于 MQTT Message 無(wú)法直接寫入 InfluxDB, InfluxDB Backend 提供了 emqx_backend_influxdb.tmpl 模板文件將 MQTT Message 轉(zhuǎn)換為可寫入 InfluxDB 的 DataPoint。

消息模板功能需要重啟 EMQ X 才能應(yīng)用更改。

tmpl 文件位于 data/templates/emqx_backend_influxdb_example.tmpl，使用 json 格式, 用戶可以為不同 Topic 定義不同的 Template, 類似:

{
    "timestamp": <Where is value of timestamp>
		"measurement": <Where is value of measurement>,
    "tags": {
        <Tag Key>: <Where is value of tag>
    },
		"fields": {
    	<Field Key>: <Where is value of field>
    }
}

其中, measurement 與 fields 為必選項(xiàng), tags 與 timestamp 為可選項(xiàng)。<Where is value of> 支持通過(guò)占位符如 $key 提取變量名為 key 的變量，支持的變量如下：

• qos: 消息 QoS

• form: 發(fā)布者信息

• topic: 發(fā)布主題

• timestamp: 時(shí)間戳

• payload.*: JSON 消息體內(nèi)任意變量，如 { "data": [{ "temp": 1 }] } 使用 ["$payload", "data", "temp"] 可以提取出 1 來(lái)

本示例設(shè)定模板如下：

{
    "sample": {
        "measurement": "$topic",
        "tags": {
            "host": ["$payload", "data", "$0", "host"],
            "region": ["$payload", "data", "$0", "region"],
            "qos": "$qos",
            "from": "$from"
        },
        "fields": {
            "temperature": ["$payload", "data", "$0", "temp"]
        },
        "timestamp": "$timestamp"
    }
}

當(dāng) Topic 為 "sample" 的 MQTT Message 擁有以下 Payload 時(shí):

{
  "data": [
    {
      "temp": 1,
      "host": "serverA",
      "region": "hangzhou"
    },
    {
      "temp": 2,
      "host": "serverB",
      "region": "ningbo"
    }
  ]
}

Backend 會(huì)將 MQTT Message 轉(zhuǎn)換為:

[
  {
    "measurement": "sample",
    "tags": {
      "from": "mqttjs_ebcc36079a",
      "host": "serverA",
      "qos": "0",
      "region": "hangzhou"
    },
    "fields": {
      "temperature": "1"
    },
    "timestamp": "1560743513626681000"
  },
  {
    "measurement": "sample",
    "tags": {
      "from": "mqttjs_ebcc36079a",
      "host": "serverB",
      "qos": "0",
      "region": "ningbo"
    },
    "fields": {
      "temperature": "2"
    },
    "timestamp": "1560743513626681000"
  }
]

使用示例

EMQ X 管理控制臺(tái) WebSocket 頁(yè)面中，向 sample 主題發(fā)布如上格式消息消息，消息將解析存儲(chǔ)到 InfluxDB udp 數(shù)據(jù)庫(kù)對(duì)應(yīng)的 measurement 中。

到此，相信大家對(duì)“如何通過(guò)InfluxDB來(lái)存儲(chǔ)相關(guān)的信息”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！