MySQL 数据存储

本章节以在 CentOS 7.2 中的实际例子来说明如何通过 MySQL 来存储相关的信息。

安装与验证 MySQL 服务器

读者可以参考 MySQL 官方文档Docker 来下载安装 MySQL 服务器,本文章使用 MySQL 5.6 版本。

为方便管理操作,可下载使用官方免费图形化管理软件 MySQL Workbeanch

配置 EMQ X 服务器

通过 RPM 方式安装的 EMQ X,MySQL 相关的配置文件位于 /etc/emqx/plugins/emqx_backend_mysql.conf,如果只是测试 MySQL 持久化的功能,大部分配置不需要做更改。填入用户名、密码、数据库即可:

  1. auth.mysql.server = 127.0.0.1:3306
  2. auth.mysql.username = root
  3. auth.mysql.password = 123456
  4. auth.mysql.database = mqtt

保持剩下部分的配置文件不变,然后需要启动该插件。启动插件的方式有 命令行控制台两种方式,读者可以任选其一。

初始化数据表

插件启动时会检查配置中数据库中是否存在以下几张数据表,初始化之后才能启动插件:

mqtt_client 存储设备在线状态

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

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

  1. DROP TABLE IF EXISTS `mqtt_sub`;
  2. CREATE TABLE `mqtt_sub` (
  3. `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  4. `clientid` varchar(64) DEFAULT NULL,
  5. `topic` varchar(255) DEFAULT NULL,
  6. `qos` int(3) DEFAULT NULL,
  7. `created` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  8. PRIMARY KEY (`id`),
  9. KEY `mqtt_sub_idx` (`clientid`,`topic`(255),`qos`),
  10. UNIQUE KEY `mqtt_sub_key` (`clientid`,`topic`)
  11. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_msg 存储 MQTT 消息

  1. DROP TABLE IF EXISTS `mqtt_msg`;
  2. CREATE TABLE `mqtt_msg` (
  3. `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  4. `msgid` varchar(100) DEFAULT NULL,
  5. `topic` varchar(1024) NOT NULL,
  6. `sender` varchar(1024) DEFAULT NULL,
  7. `node` varchar(60) DEFAULT NULL,
  8. `qos` int(11) NOT NULL DEFAULT '0',
  9. `retain` tinyint(2) DEFAULT NULL,
  10. `payload` blob,
  11. `arrived` datetime NOT NULL, -- 是否抵达(QoS > 0
  12. PRIMARY KEY (`id`)
  13. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_retain 存储 Retain 消息

  1. DROP TABLE IF EXISTS `mqtt_retain`;
  2. CREATE TABLE `mqtt_retain` (
  3. `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  4. `topic` varchar(200) DEFAULT NULL,
  5. `msgid` varchar(60) DEFAULT NULL,
  6. `sender` varchar(100) DEFAULT NULL,
  7. `node` varchar(100) DEFAULT NULL,
  8. `qos` int(2) DEFAULT NULL,
  9. `payload` blob,
  10. `arrived` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  11. PRIMARY KEY (`id`),
  12. UNIQUE KEY `mqtt_retain_key` (`topic`)
  13. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_acked 存储客户端消息确认

  1. DROP TABLE IF EXISTS `mqtt_acked`;
  2. CREATE TABLE `mqtt_acked` (
  3. `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  4. `clientid` varchar(200) DEFAULT NULL,
  5. `topic` varchar(200) DEFAULT NULL,
  6. `mid` int(200) DEFAULT NULL,
  7. `created` timestamp NULL DEFAULT NULL,
  8. PRIMARY KEY (`id`),
  9. UNIQUE KEY `mqtt_acked_key` (`clientid`,`topic`)
  10. ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

通过命令行启动

  1. emqx_ctl plugins load emqx_backend_mysql

通过管理控制台启动

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

客户端在线状态存储

客户端上下线时,插件将更新在线状态、上下线时间、节点客户端列表至 MySQL 数据库。

配置项

打开配置文件,配置 Backend 规则:

  1. ## hook: client.connected、client.disconnected
  2. ## action/function: on_client_connected、on_client_disconnected
  3. ## 客户端上下线
  4. backend.mysql.hook.client.connected.1 = {"action": {"function": "on_client_connected"}, "pool": "pool1"}
  5. backend.mysql.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}

使用示例

浏览器打开 http://127.0.0.1:18083 EMQ X 管理控制台,在 工具 -> Websocket 中新建一个客户端连接,指定 clientid 为 sub_client,点击连接,连接成功后手动断开:

image-20181116105333637

在 MySQL Workbeanch 中点击 mqtt_client 表查看,此时将写入 / 更新一条客户端上下线记录:

image-20181119105034528

客户端代理订阅

客户端上线时,存储模块直接从数据库读取预设待订阅列表,代理加载订阅主题。在客户端需要通过预定主题通信(接收消息)场景下,应用能从数据层面设定 / 改变代理订阅列表。

配置项

打开配置文件,配置 Backend 规则:

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

使用示例

sub_client 设备上线时,需要为其订阅 sub_client/upstreamsub_client/downlink 两个 QoS 1 的主题:

  1. mqtt_sub 表中初始化插入代理订阅主题信息:
  1. insert into mqtt_sub(clientid, topic, qos) values("sub_client", "sub_client/upstream", 1);
  2. insert into mqtt_sub(clientid, topic, qos) values("sub_client", "sub_client/downlink", 1);
  1. EMQ X 管理控制台 WebSocket 页面,以 clientid sub_client 新建一个客户端连接,切换至订阅页面,可见当前客户端自动订阅了 sub_client/upstreamsub_client/downlink 两个 QoS 1 的主题:

image-20181116110036523

  1. 切换回管理控制台 WebSocket 页面,向 sub_client/downlink 主题发布消息,可在消息订阅列表收到发布的消息。

持久化发布消息

配置项

打开配置文件,配置 Backend 规则,支持使用 topic 参数进行消息过滤,此处使用 # 通配符存储任意主题消息:

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

使用示例

在 EMQ X 管理控制台 WebSocket 页面中,使用 clientdi sub_client 建立连接,向主题 upstream_topic 发布多条消息,EMQ X 将消息列表持久化至 mqtt_msg 表中:

image-20181119110712267

暂只支持 QoS 1 2 的消息持久化。

Retain 消息持久化

配置项

打开配置文件,配置 Backend 规则:

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

使用示例

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

image-20181119111926675

发布(消息不为空)

非空的 retain 消息发布时,EMQ X 将以 topic 为唯一键,持久化该条消息至 mqtt_retain 表中,相同主题下发从不同的 retain 消息,只有最后一条消息会被持久化:

image-20181119164153931

订阅

客户端订阅 retain 主题后,EMQ X 将查询 mqtt_retain 数据表,执行投递 retain 消息操作。

发布(消息为空)

MQTT 协议中,发布空的 retain 消息将清空 retain 记录,此时 retain 记录将从 mqtt_retain 表中删除。

消息确认持久化

开启消息确认 (ACK) 持久化后,客户端订阅 QoS 1、QoS 2 级别的主题时,EMQ X 将在数据库以 clientid + topic 为唯一键初始化 ACK 记录。

配置项

打开配置文件,配置 Backend 规则,可使用 topic 通配符 过滤要应用的消息:

  1. ## 订阅时初始化 ACK 记录
  2. backend.mysql.hook.session.subscribed.1 = {"topic": "#", "action": {"function": "on_message_fetch"}, "pool": "pool1"}
  3. ## 消息抵达时更新抵达状态
  4. backend.mysql.hook.message.acked.1 = {"topic": "#", "action": {"function": "on_message_acked"}, "pool": "pool1"}
  5. ## 取消订阅时删除记录行
  6. 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 的主题:

image-20181119140251843

此时 mqtt_acked 表将插入初始化数据行,每向主题发布一条 QoS > 0 的消息,消息抵达后数据行 mid 将自增 1:

image-20181119140354855

代理订阅中满足 QoS > 0 的 topic 也会初始化记录,客户端取消订阅后相关记录将被删除。

自定义 SQL

除去插件内置函数、表结构外,emqx_backend_mysql 还支持自定义 SQL 语句,通过使用如 ${clientid} 模板语法动态构造 SQL 语句实现如客户端连接历史、更新自定义数据表等操作。

SQL语句参数说明

hook可用参数示例(sql语句中${name} 表示可获取的参数)
client.connectedclientidinsert into conn(clientid) values(${clientid})
client.disconnectedclientidinsert into disconn(clientid) values(${clientid})
session.subscribedclientid, topic, qosinsert into sub(topic, qos) values(${topic}, ${qos})
session.unsubscribedclientid, topicdelete from sub where topic = ${topic}
message.publishmsgid, topic, payload, qos, clientidinsert into msg(msgid, topic) values({msgid}, ${topic})
message.ackedmsgid, topic, clientidinsert into ack(msgid, topic) values({msgid}, ${topic})
message.deliveredmsgid, topic, clientidinsert into delivered(msgid, topic) values({msgid}, ${topic})

客户端连接 log 示例

设计表结构如下:

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

自定义 SQL:

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

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

image-20181119154828728

高级选项

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

总结

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