MySQL 数据存储

配置文件: emqx_backend_mysql.conf

配置 MySQL 服务器

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

  1. ## Mysql 服务器地址
  2. backend.mysql.pool1.server = 127.0.0.1:3306
  3. ## Mysql 连接池大小
  4. backend.mysql.pool1.pool_size = 8
  5. ## Mysql 用户名
  6. backend.mysql.pool1.user = root
  7. ## Mysql 密码
  8. backend.mysql.pool1.password = public
  9. ## Mysql 数据库名称
  10. 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"}
  11. ## 获取离线消息
  12. ## "offline_opts": 获取离线消息的配置
  13. ## - max_returned_count: 单次拉去的最大离线消息数目
  14. ## - time_range: 仅拉去在当前时间范围的消息
  15. ## backend.mysql.hook.session.subscribed.1 = {"topic": "#", "action": {"function": "on_message_fetch"}, "offline_opts": {"max_returned_count": 500, "time_range": "2h"}, "pool": "pool1"}
  16. ## 如果需要存储 Qos0 消息, 可开启以下配置
  17. ## 警告: 当开启以下配置时, 需关闭 'on_message_fetch', 否则 qos1, qos2 消息会被存储俩次
  18. ## 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";
  2. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  3. | id | clientid | state | node | online_at | offline_at | created |
  4. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  5. | 1 | test | 1 | emqx@127.0.0.1 | 2016-11-15 09:40:40 | NULL | 2016-12-24 09:40:22 |
  6. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  7. 1 rows in set (0.00 sec)

例如 ClientId 为 test 客户端下线:

  1. select * from mqtt_client where clientid = "test";
  2. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  3. | id | clientid | state | node | online_at | offline_at | created |
  4. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  5. | 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 |
  6. +----+----------+-------+----------------+---------------------+---------------------+---------------------+
  7. 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";
  2. +----+--------------+-------------+------+---------------------+
  3. | id | clientId | topic | qos | created |
  4. +----+--------------+-------------+------+---------------------+
  5. | 1 | test | test_topic1 | 1 | 2016-12-24 17:09:05 |
  6. | 2 | test | test_topic2 | 2 | 2016-12-24 17:12:51 |
  7. +----+--------------+-------------+------+---------------------+
  8. 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";
  2. +----+-------------------------------+----------+--------+------+-----+--------+---------+---------------------+
  3. | id | msgid | topic | sender | node | qos | retain | payload | arrived |
  4. +----+-------------------------------+----------+--------+------+-----+--------+---------+---------------------+
  5. | 1 | 53F98F80F66017005000004A60003 | hello | test | NULL | 1 | 0 | hello | 2016-12-24 17:25:12 |
  6. | 2 | 53F98F9FE42AD7005000004A60004 | world | test | NULL | 1 | 0 | world | 2016-12-24 17:25:45 |
  7. +----+-------------------------------+----------+--------+------+-----+--------+---------+---------------------+
  8. 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";
  2. +----+----------+-------------------------------+---------+------+------+---------+---------------------+
  3. | id | topic | msgid | sender | node | qos | payload | arrived |
  4. +----+----------+-------------------------------+---------+------+------+---------+---------------------+
  5. | 1 | retain | 53F33F7E4741E7007000004B70001 | test | NULL | 1 | www | 2016-12-24 16:55:18 |
  6. +----+----------+-------------------------------+---------+------+------+---------+---------------------+
  7. > 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