EMQ X 插件持久化系列 （五）MySQL MQTT 数据存储

小伙伴们有没有觉得学习数据库很有意思？有意思就对了！今天就给大家带来《EMQ X 插件持久化系列 （五）MySQL MQTT 数据存储》，以下内容将会涉及到MySQL、持久化、物联网、iot，若是在学习中对其中部分知识点有疑问，或许看了本文就能帮到你！

本文以在

DROP TABLE IF EXISTS `mqtt_client`;
CREATE TABLE `mqtt_client` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `clientid` varchar(64) DEFAULT NULL,
  `state` varchar(3) DEFAULT NULL, -- 在线状态 0 离线 1 在线
  `node` varchar(100) DEFAULT NULL, -- 所属节点
  `online_at` datetime DEFAULT NULL, -- 上线时间
  `offline_at` datetime DEFAULT NULL, -- 下线时间
  `created` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `mqtt_client_idx` (`clientid`),
  UNIQUE KEY `mqtt_client_key` (`clientid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_sub 存储设备的主题订阅关系

DROP TABLE IF EXISTS `mqtt_sub`;
CREATE TABLE `mqtt_sub` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `clientid` varchar(64) DEFAULT NULL,
  `topic` varchar(255) DEFAULT NULL,
  `qos` int(3) DEFAULT NULL,
  `created` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `mqtt_sub_idx` (`clientid`,`topic`(255),`qos`),
  UNIQUE KEY `mqtt_sub_key` (`clientid`,`topic`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_msg 存储 MQTT 消息

DROP TABLE IF EXISTS `mqtt_msg`;
CREATE TABLE `mqtt_msg` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `msgid` varchar(100) DEFAULT NULL,
  `topic` varchar(1024) NOT NULL,
  `sender` varchar(1024) DEFAULT NULL,
  `node` varchar(60) DEFAULT NULL,
  `qos` int(11) NOT NULL DEFAULT '0',
  `retain` tinyint(2) DEFAULT NULL,
  `payload` blob,
  `arrived` datetime NOT NULL, -- 是否抵达（QoS > 0）
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_retain 存储 Retain 消息

DROP TABLE IF EXISTS `mqtt_retain`;
CREATE TABLE `mqtt_retain` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `topic` varchar(200) DEFAULT NULL,
  `msgid` varchar(60) DEFAULT NULL,
  `sender` varchar(100) DEFAULT NULL,
  `node` varchar(100) DEFAULT NULL,
  `qos` int(2) DEFAULT NULL,
  `payload` blob,
  `arrived` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `mqtt_retain_key` (`topic`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_acked 存储客户端消息确认

DROP TABLE IF EXISTS `mqtt_acked`;
CREATE TABLE `mqtt_acked` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `clientid` varchar(200) DEFAULT NULL,
  `topic` varchar(200) DEFAULT NULL,
  `mid` int(200) DEFAULT NULL,
  `created` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `mqtt_acked_key` (`clientid`,`topic`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

配置 EMQ X 服务器

通过 RPM 方式安装的 EMQ X，MySQL 相关的配置文件位于

auth.mysql.server = 127.0.0.1:3306

auth.mysql.username = root

auth.mysql.password = 123456

auth.mysql.database = mqtt

保持剩下部分的配置文件不变，然后需要启动该插件。启动插件的方式有

emqx_ctl plugins load emqx_backend_mysql

通过管理控制台启动

EMQ X 管理控制台 插件 页面中，找到 emqx_backend_mysql 插件，点击 启动。

通过 REST API 启动

使用

## hook: client.connected、client.disconnected
## action/function: on_client_connected、on_client_disconnected


## 客户端上下线
backend.mysql.hook.client.connected.1 = {"action": {"function": "on_client_connected"}, "pool": "pool1"}

backend.mysql.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}

使用示例

浏览器打开

## hook: client.connected
## action/function: on_subscribe_lookup
backend.mysql.hook.client.connected.2    = {"action": {"function": "on_subscribe_lookup"}, "pool": "pool1"}

使用示例

当

insert into mqtt_sub(clientid, topic, qos) values("sub_client", "sub_client/upstream", 1);
insert into mqtt_sub(clientid, topic, qos) values("sub_client", "sub_client/downlink", 1);

1. EMQ X 管理控制台 WebSocket 页面，以 clientid

   ## hook: message.publish
   ## action/function: on_message_publish
   
   backend.mysql.hook.message.publish.1     = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}

   使用示例

   在 EMQ X 管理控制台 WebSocket 页面中，向主题

   ## 同时开启以下规则，启用 retain 持久化三个生命周期
   
   ## 发布非空 retain 消息时 (存储)
   backend.mysql.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "pool": "pool1"}
   
   ## 设备订阅主题时查询 retain 消息
   backend.mysql.hook.session.subscribed.2  = {"topic": "#", "action": {"function": "on_retain_lookup"}, "pool": "pool1"}
   
   ## 发布空 retain 消息时 (清除)
   backend.mysql.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}
   

   使用示例

   在 EMQ X 管理控制台 WebSocket 页面中建立连接后，发布消息勾选保留：

   

   发布（消息不为空）

   非空的 retain 消息发布时，EMQ X 将以 topic 为唯一键，持久化该条消息至

   ## 订阅时初始化 ACK 记录
   backend.mysql.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch"}, "pool": "pool1"}
   
   
   ## 消息抵达时更新抵达状态
   backend.mysql.hook.message.acked.1       = {"topic": "#", "action": {"function": "on_message_acked"}, "pool": "pool1"}
   
   ## 取消订阅时删除记录行
   backend.mysql.hook.session.unsubscribed.1= {"topic": "#", "action": {"sql": ["delete from mqtt_acked where clientid = ${clientid} and topic = ${topic}"]}, "pool": "pool1"}

   使用示例

   在 EMQ X 管理控制台 WebSocket 页面中建立连接后，订阅 QoS > 0 的主题：

   

   此时

   CREATE TABLE `mqtt`.`connect_logs` (
     `id` INT NOT NULL AUTO_INCREMENT,
     `clientid` VARCHAR(255) NULL,
     `created_at` TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP, -- 记录时间
     `state` INT NOT NULL DEFAULT 0,  -- 记录类型： 0 下线 1 上线
     PRIMARY KEY (`id`)
   ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

   自定义 SQL：

   ## connected hook 中配置自定义 SQL
   ## 可以配置多条 SQL 语句 "SQL": ["sql_a", "sql_b", "sql_c"]
   
   ## 连接时
   backend.mysql.hook.client.connected.3 = {"action": {"sql": ["insert into connect_logs(clientid, state) values(${clientid}, 1)"]}, "pool": "pool1"}
   
   ## 断开时
   backend.mysql.hook.client.disconnected.3 = {"action": {"sql": ["insert into connect_logs(clientid, state) values(${clientid}, 0)"]}, "pool": "pool1"}

   客户端上下线时将填充并执行预定的 SQL 语句，将连接记录写入

   backend.mysql.time_range = 5s
   
   backend.mysql.max_returned_count = 500

   总结

   读者在理解了 MySQL 中所存储的数据结构、自定义 SQL 之后，可以结合 MySQL 拓展相关应用。

   版权声明： 本文为 EMQ 原创，转载请注明出处。

   原文链接：https://www.emqx.io/cn/blog/emqx-plugin-persistence-series-5-mysql-mqtt-data-storage

   文中关于mysql的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《EMQ X 插件持久化系列 （五）MySQL MQTT 数据存储》文章吧，也可关注golang学习网公众号了解相关技术文章。