关于java:场景三将-EMQ-X-指定主题下且满足条件的消息存储到-Mysql

场景介绍

该场景须要将 EMQ X 指定主题下且满足条件的音讯存储到 MySQL 数据库。为了便于后续剖析检索，音讯内容须要进行拆分存储。
该场景下设施端上报信息如下：

• 上报主题：cmd/state/:id，主题中 id 代表车辆客户端识别码

• 音讯体：

  {
    "id": "NXP-058659730253-963945118132721-22", // 客户端识别码
    "speed": 32.12, // 车辆速度
    "direction": 198.33212, // 行驶方向
    "tachometer": 3211, // 发动机转速，数值大于 8000 时才需存储
    "dynamical": 8.93, // 刹时油耗
    "location": { // GPS 经纬度数据
      "lng": 116.296011,
      "lat": 40.005091
    },
    "ts": 1563268202 // 上报工夫
  } 

当上报数据发动机转速数值大于 8000 时，存储以后信息以便后续剖析用户车辆应用状况。

筹备工作

创立数据库

创立 mybatis_plus 数据库以存储音讯数据，这里指定数据库编码为 utf8mb4 防止编码问题：

CREATE DATABASE `mybatis_plus` CHARACTER SET utf8mb4;

创立数据表

依据场景需要，创立数据表 use_statistics 构造及字段正文如下：

CREATE TABLE `use_statistics` (`id` int(11) NOT NULL AUTO_INCREMENT,
  `client_id` varchar(100) DEFAULT NULL COMMENT '客户端识别码',
  `speed` float unsigned DEFAULT '0.00' COMMENT '以后车速',
  `tachometer` int(11) unsigned DEFAULT '0' COMMENT '发动机转速',
  `ts` int(11) unsigned DEFAULT '0' COMMENT '上报工夫戳',
  `msg_id` varchar(50) DEFAULT NULL COMMENT 'MQTT 音讯 ID',
  PRIMARY KEY (`id`),
  KEY `client_id_index` (`client_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

创立胜利后能够看到新建的数据表：

配置阐明

创立资源

拜访 http://ip:18083/#/monitor 关上 EMQ X Dashboard，进入左侧菜单的 资源 页面，点击 创立 按钮，键入 MySQL 服务器信息进行资源创立。

创立胜利，咱们能够在 资源 页面看到咱们创立好的资源

资源创立胜利后点击列表中 状态按钮 ，查看各个节点资源连贯情况，如果节点上资源不可用，请查看配置是否正确、网络连通性，并点击 重连 按钮手动重连。

节点 绿色 为可用状态

创立规定

进入左侧菜单的 规定 页面，点击 创立 按钮，进行规定创立。咱们可在界面上看到可用字段及示例 SQL：

筛选所需字段

规定引擎应用 SQL 语句解决规定条件，该业务中咱们须要将 payload 中所有字段独自抉择进去，应用 payload.fieldName 格局进行抉择，还须要音讯上下文的 topic、qos、id 信息，以后 SQL 如下：

SELECT

  payload.id AS client_id,    payload.speed AS speed,    payload.tachometer AS tachometer,    payload.ts AS ts,    id

FROM

  "#"

确立筛选条件

应用 SQL 语句 WHERE 字句进行条件筛选，该业务中咱们须要定义两个条件：

• 仅解决 cmd/state/:id 主题，应用主题通配符 =~ 对 topic 进行筛选：topic =~ 'cmd/state/+'
• 仅解决 tachometer > 8000 的音讯，应用比拟符对 tachometer 进行筛选：payload.tachometer > 8000

组合上一步骤失去 SQL 如下：

SELECT

  payload.id AS client_id,    payload.speed AS speed,    payload.tachometer AS tachometer,    payload.ts AS ts,    id

FROM

  "#"
  WHERE
  topic =~ 'cmd/state/+'
  AND payload.tachometer > 8000

应用 SQL 测试性能进行输入测试

借助 SQL 测试性能，咱们能够实时查看以后 SQL 解决后的数据输入，该性能须要咱们指定 payload 等模仿原始数据。

payload 数据如下，留神更改 tachometer 数值大小，以满足 SQL 条件：

{
  "id": "NXP-058659730253-963945118132721-22",
  "speed": 32.12,
  "direction": 198.33212,
  "tachometer": 9001,
  "dynamical": 8.93,
  "location": {
    "lng": 116.296011,
    "lat": 40.005091
  },
  "ts": 1563268202
}

点击 SQL 测试 切换按钮，更改 topic 与 payload 为场景中的信息，点击 测试 按钮查看数据输入：

测试输入数据为：

{
  "ts": 1563268202,
  "tachometer": 9001,
  "speed": 32.12,
  "id": "5B5860A7FF023F44310005D810000",
  "client_id": "NXP-058659730253-963945118132721-22"
}

测试输入与预期相符，咱们能够进行后续步骤。

增加响应动作，存储音讯到 MySQL

SQL 条件输入输出无误后，咱们持续增加响应动作，配置写入 SQL 语句，将筛选后果存储到 MySQL。

点击响应动作中的 增加 按钮，抉择 保留数据到 MySQL 动作，选取刚刚选定的资源，咱们应用 ${fieldName} 语法填充 SQL 语句，将数据插入到数据库，最初点击 新建 按钮实现规定创立。

动作的 SQL 配置如下：

INSERT INTO 
    `use_statistics` (`client_id`, `speed`, `tachometer`, `ts`, `msg_id`)
VALUES 
    (${client_id}, ${speed}, ${tachometer}, ${ts}, ${id});

测试

预期后果

咱们胜利创立了一条规定，蕴含一个解决动作，动作冀望成果如下：

1. 设施向 cmd/state/:id 主题上报音讯是，当音讯中的 tachometer 数值超过 8000 时将命中 SQL，规定列表中 已命中 数字减少 1；
2. MySQL iot_data数据库的 use_statistics 表中将减少一条数据，数值与以后音讯统一。

应用 Dashboard 中的 Websocket 工具测试

切换到 工具 –> Websocket 页面，应用任意信息客户端连贯到 EMQ X，连贯胜利后在 音讯 卡片发送如下信息：

• 主题：cmd/state/NXP-058659730253-963945118132721-22

• 音讯体：

  {
    "id": "NXP-058659730253-963945118132721-22",
    "speed": 32.12,
    "direction": 198.33212,
    "tachometer": 9002,
    "dynamical": 8.93,
    "location": {
      "lng": 116.296011,
      "lat": 40.005091
    },
    "ts": 1563268202
  }

点击 发送 按钮，发送胜利后查看规定统计 已命中 数据统计值为 1 表明规定已胜利命中

MySQL 命令行中查看数据表记录失去数据如下：

至此，咱们通过规定引擎实现了应用规定引擎存储音讯到 MySQL 数据库的业务开发。
返回上一级 场景二：将 EMQ X 指定主题下且满足条件的音讯存储到 Reids