保存数据到 InfluxDB

搭建 InfluxDB 数据库环境,以 macOS X 为例:

  1. $ docker pull influxdb
  2. $ docker run --name=influxdb --rm -d -p 8086:8086 -p 8089:8089/udp -v ${PWD}/files/influxdb.conf:/etc/influxdb/influxdb.conf influxdb:latest

EMQ X 仅支持通过 UDP 协议连接 InfluxDB,需要修改默认 InfluxDB 配置文件如下:

  1. [[udp]]
  2. enabled = true
  3. bind-address = ":8089"
  4. # 消息保存的数据库
  5. database = "emqx"
  6. # InfluxDB precision for timestamps on received points ("" or "n", "u", "ms", "s", "m", "h")
  7. # EMQ X 默认时间戳是毫秒
  8. precision = "ms"
  9. # 其他配置根据需要自行修改
  10. # batch-size = 1000
  11. # batch-pending = 5
  12. # batch-timeout = "5s"
  13. # read-buffer = 1024

创建规则:

打开 emqx dashboard,选择左侧的 “规则” 选项卡。

选择触发事件 “消息发布”,然后填写规则 SQL:

  1. SELECT
  2. payload as p,
  3. p.host as host,
  4. p.location as location,
  5. p.internal as internal,
  6. p.external as external
  7. FROM
  8. "#"

image

关联动作:

在 “响应动作” 界面选择 “添加”,然后在 “动作” 下拉框里选择 “保存数据到 InfluxDB”。

image

填写动作参数:

“保存数据到 InfluxDB” 动作需要六个参数:

1). Measurement。指定写入到 InfluxDB 的 data point 的 measurement。

2). Field Keys。指定写入到 InfluxDB 的 data point 的 fields 的值从哪里获取。

3). Tags Keys。指定写入到 InfluxDB 的 data point 的 tags 的值从哪里获取。

4). Timestamp Key。指定写入到 InfluxDB 的 data point 的 timestamp 的值从哪里获取。

5). 设置时间戳。未指定 Timestamp Key 时是否自动生成。

6). 关联资源。现在资源下拉框为空,可以点击右上角的 “新建资源” 来创建一个 InfluxDB 资源:

image

选择 “InfluxDB 资源”:

image

填写资源配置:

本示例中所有配置保持默认值即可,点击 “测试连接” 按钮,确保连接测试成功。

最后点击 “新建” 按钮。

image

返回响应动作界面,点击 “确认”。 返回规则创建界面,点击 “新建”。

image

规则已经创建完成,现在发一条消息:

  1. Topic: "t/1"
  2. QoS: 0
  3. Retained: false
  4. Payload:
  5. "{"host":"serverA","location":"roomA","internal":25,"external":37}"

然后检查 InfluxDB,新的 data point 是否添加成功:

  1. $ docker exec -it influxdb influx
  2. use db
  3. Using database db
  4. select * from "temperature"
  5. name: temperature
  6. time external host internal location
  7. ---- -------- ---- -------- --------
  8. 1561535778444457348 35 serverA 25 roomA

在规则列表里,可以看到刚才创建的规则的命中次数已经增加了 1:

image