MySQL 数据存储

配置文件: emqx_backend_mysql.conf

配置 MySQL 服务器

支持配置多台 MySQL 服务器连接池:

  1. ## Mysql 服务器地址
  2. backend.mysql.pool1.server = 127.0.0.1:3306
  4. ## Mysql 连接池大小
  5. backend.mysql.pool1.pool_size = 8
  7. ## Mysql 用户名
  8. backend.mysql.pool1.user = root
  10. ## Mysql 密码
  11. backend.mysql.pool1.password = public
  13. ## Mysql 数据库名称
  14. backend.mysql.pool1.database = mqtt

配置 MySQL 存储规则

  1. backend.mysql.hook.client.connected.1    = {"action": {"function": "on_client_connected"}, "pool": "pool1"}
  2. backend.mysql.hook.session.created.1     = {"action": {"function": "on_subscribe_lookup"}, "pool": "pool1"}
  3. backend.mysql.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}
  4. backend.mysql.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch"}, "pool": "pool1"}
  5. backend.mysql.hook.session.subscribed.2  = {"topic": "#", "action": {"function": "on_retain_lookup"}, "pool": "pool1"}
  6. backend.mysql.hook.session.unsubscribed.1= {"topic": "#", "action": {"sql": ["delete from mqtt_acked where clientid = ${clientid} and topic = ${topic}"]}, "pool": "pool1"}
  7. backend.mysql.hook.message.publish.1     = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}
  8. backend.mysql.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "pool": "pool1"}
  9. backend.mysql.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}
  10. backend.mysql.hook.message.acked.1       = {"topic": "#", "action": {"function": "on_message_acked"}, "pool": "pool1"}
  12. ## 获取离线消息
  13. ##  "offline_opts": 获取离线消息的配置
  14. ##     - max_returned_count: 单次拉去的最大离线消息数目
  15. ##     - time_range: 仅拉去在当前时间范围的消息
  16. ## backend.mysql.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch"}, "offline_opts": {"max_returned_count": 500, "time_range": "2h"}, "pool": "pool1"}
  18. ## 如果需要存储 Qos0 消息, 可开启以下配置
  19. ## 警告: 当开启以下配置时, 需关闭 'on_message_fetch', 否则 qos1, qos2 消息会被存储俩次
  20. ## backend.mysql.hook.message.publish.4     = {"topic": "#", "action": {"function": "on_message_store"}, "pool": "pool1"}

MySQL 存储规则说明

hooktopicaction说明
client.connectedon_client_connected存储客户端在线状态
session.createdon_subscribe_lookup订阅主题
client.disconnectedon_client_disconnected存储客户端离线状态
session.subscribed#on_message_fetch获取离线消息
session.subscribed#on_retain_lookup获取retain消息
message.publish#on_message_publish存储发布消息
message.publish#on_message_retain存储retain消息
message.publish#on_retain_delete删除retain消息
message.acked#on_message_acked消息ACK处理

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.delivermsgid, topic, clientidinsert into deliver(msgid, topic) values(${msgid}, ${topic})

SQL 语句配置 Action

MySQL 存储支持用户采用 SQL 语句配置 Action:

  1. ## 在客户端连接到 EMQ X 服务器后，执行一条 sql 语句(支持多条 sql 语句)
  2. backend.mysql.hook.client.connected.3 = {"action": {"sql": ["insert into conn(clientid) values(${clientid})"]}, "pool": "pool1"}

创建 MySQL 数据库表

  1. create database mqtt;

导入 MySQL 库表结构

  1. mysql -u root -p mqtt < etc/sql/emqx_backend_mysql.sql

数据库名称可自定义

MySQL 设备在线状态表

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,
  6.   `node` varchar(64) 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.   INDEX topic_index(`id`, `clientid`)
  14. ) ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;

查询设备在线状态:

  1. select * from mqtt_client where clientid = ${clientid};

例如 ClientId 为 test 客户端上线:

  1.  select * from mqtt_client where clientid = "test";
  3. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  4. | id | clientid | state | node           | online_at           | offline_at          | created             |
  5. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  6.    |  1 | test     | 1     | emqx@127.0.0.1 | 2016-11-15 09:40:40 | NULL                | 2016-12-24 09:40:22 |
  7. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  8. 1 rows in set (0.00 sec)

例如 ClientId 为 test 客户端下线:

  1. select * from mqtt_client where clientid = "test";
  3. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  4. | id | clientid | state | node           | online_at           | offline_at          | created             |
  5. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  6. |  1 | test     | 0     | emqx@127.0.0.1 | 2016-11-15 09:40:40 | 2016-11-15 09:46:10 | 2016-12-24 09:40:22 |
  7. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  8. 1 rows in set (0.00 sec)

MySQL 主题订阅表

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(180) DEFAULT NULL,
  6.   `qos` tinyint(1) DEFAULT NULL,
  7.   `created` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  8.   PRIMARY KEY (`id`),
  9.   KEY `mqtt_sub_idx` (`clientid`,`topic`,`qos`),
  10.   UNIQUE KEY `mqtt_sub_key` (`clientid`,`topic`),
  11.   INDEX topic_index(`id`, `topic`)
  12. ) ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;

例如 ClientId 为 test 客户端订阅主题 test_topic1 test_topic2:

  1. insert into mqtt_sub(clientid, topic, qos) values("test", "test_topic1", 1);
  2. insert into mqtt_sub(clientid, topic, qos) values("test", "test_topic2", 2);

某个客户端订阅主题:

  1.  select * from mqtt_sub where clientid = ${clientid};

查询 ClientId 为 test 的客户端已订阅主题:

  1. select * from mqtt_sub where clientid = "test";
  3. +----+--------------+-------------+------+---------------------+
  4. | id | clientId     | topic       | qos  | created             |
  5. +----+--------------+-------------+------+---------------------+
  6. |  1 | test         | test_topic1 |    1 | 2016-12-24 17:09:05 |
  7. |  2 | test         | test_topic2 |    2 | 2016-12-24 17:12:51 |
  8. +----+--------------+-------------+------+---------------------+
  9. 2 rows in set (0.00 sec)

MySQL 消息存储表

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(64) DEFAULT NULL,
  5.   `topic` varchar(180) NOT NULL,
  6.   `sender` varchar(64) DEFAULT NULL,
  7.   `node` varchar(64) DEFAULT NULL,
  8.   `qos` tinyint(1) NOT NULL DEFAULT '0',
  9.   `retain` tinyint(1) DEFAULT NULL,
  10.   `payload` blob,
  11.   `arrived` datetime NOT NULL,
  12.   PRIMARY KEY (`id`),
  13.   INDEX topic_index(`id`, `topic`)
  14. ) ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;

查询某个客户端发布的消息:

  1. select * from mqtt_msg where sender = ${clientid};

查询 ClientId 为 test 的客户端发布的消息:

  1. select * from mqtt_msg where sender = "test";
  3. +----+-------------------------------+----------+--------+------+-----+--------+---------+---------------------+
  4. | id | msgid                         | topic    | sender | node | qos | retain | payload | arrived             |
  5. +----+-------------------------------+----------+--------+------+-----+--------+---------+---------------------+
  6. | 1  | 53F98F80F66017005000004A60003 | hello    | test   | NULL |   1 |      0 | hello   | 2016-12-24 17:25:12 |
  7. | 2  | 53F98F9FE42AD7005000004A60004 | world    | test   | NULL |   1 |      0 | world   | 2016-12-24 17:25:45 |
  8. +----+-------------------------------+----------+--------+------+-----+--------+---------+---------------------+
  9. 2 rows in set (0.00 sec)

MySQL 保留消息表

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(180) DEFAULT NULL,
  5.   `msgid` varchar(64) DEFAULT NULL,
  6.   `sender` varchar(64) DEFAULT NULL,
  7.   `node` varchar(64) DEFAULT NULL,
  8.   `qos` tinyint(1) 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.   INDEX topic_index(`id`, `topic`)
  14. ) ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;

查询 retain 消息:

  1. select * from mqtt_retain where topic = ${topic};

查询 topic 为 retain 的 retain 消息:

  1. select * from mqtt_retain where topic = "retain";
  3. +----+----------+-------------------------------+---------+------+------+---------+---------------------+
  4. | id | topic    | msgid                         | sender  | node | qos  | payload | arrived             |
  5. +----+----------+-------------------------------+---------+------+------+---------+---------------------+
  6. |  1 | retain   | 53F33F7E4741E7007000004B70001 | test    | NULL |    1 | www     | 2016-12-24 16:55:18 |
  7. +----+----------+-------------------------------+---------+------+------+---------+---------------------+
  10. >    1 rows in set (0.00 sec)

MySQL 消息确认表

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(64) DEFAULT NULL,
  5.   `topic` varchar(180) DEFAULT NULL,
  6.   `mid` int(11) unsigned DEFAULT NULL,
  7.   `created` timestamp NULL DEFAULT NULL,
  8.   PRIMARY KEY (`id`),
  9.   UNIQUE KEY `mqtt_acked_key` (`clientid`,`topic`),
  10.   INDEX topic_index(`id`, `topic`)
  11. ) ENGINE=InnoDB DEFAULT CHARSET=utf8MB4;

启用 MySQL 数据存储插件

  1. ./bin/emqx_ctl plugins load emqx_backend_mysql