EMQ X 数据持久化 存储消息到 Redis

Redis

简介

Redis 是完全开源的,遵守 BSD 协议,是一个高性能的 key-value 数据库。

Redis 与其他 key - value 缓存产品有以下三个特点:

  • Redis支持数据的持久化,可以将内存中的数据保存在磁盘中,重启的时候可以再次加载进行使用。
  • Redis不仅仅支持简单的key-value类型的数据,同时还提供list,set,zset,hash等数据结构的存储。
  • Redis支持数据的备份,即master-slave模式的数据备份。

可以参考centos8.0 安装redis
可以参考centos8.0 平台用redis-cli对redis5监控和管理
来安装 Redis,通过 redis-server 命令来启动 Redis 服务器。
如果安装的 Redis 不与 EMQ X 在同一服务器上,请指定正确的 Redis 服务器的地址与端口。通过 RPM 方式安装的 EMQ X,Redis 相关的配置文件位于/etc/emqx/plugins/emqx_backend_redis.conf如下所示:
vim emqx_backend_redis.conf

常用命令

  • 查看redis 状态:systemctl status redis
  • 启动redis:systemctl start redis
  • 重启redis:systemctl restart redis.service
  • 查看redis安装目录:ps -ef|grep redis

ps -ef|grep redis得到进程号 29895
然后:ls -l /proc/29895/cwd
EMQ X 数据持久化 存储消息到 Redis_第1张图片

启动插件

命令:emqx_ctl plugins load emqx_backend_redis
EMQ X 数据持久化 存储消息到 Redis_第2张图片

场景一:客户端在线状态存储

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

配置项

打开配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf],配置 Backend 规则:

## 上线
backend.redis.hook.client.connected.1    =  { "action": { "function": "on_client_connected" }, "pool": "pool1"}

## 下线
backend.redis.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}

EMQ X 数据持久化 存储消息到 Redis_第3张图片

使用示例

浏览器打开 http://ip:18083 EMQ X Dashboard 管理控制台,在 工具 -> Websocket 中新建一个客户端连接,指定 clientid 为 sub_client
EMQ X 数据持久化 存储消息到 Redis_第4张图片

常用命令

  • 连接列表命令:HGETALL mqtt:node:emqx@127.0.0.1

HGETALL mqtt:node:emqx@127.0.0.1
字段说明:

## 节点下在线设备信息
127.0.0.1:6379> HGETALL mqtt:node:emqx@127.0.0.1
1) "sub_client1" # clientid
2) "1542272836" # 上线时间时间戳
  • 连接详细信息命令:HGETALL mqtt:client:sub_client

EMQ X 数据持久化 存储消息到 Redis_第5张图片
字段说明:

## 客户端在线状态
127.0.0.1:6379> HGETALL mqtt:client:sub_client
1) "state"
2) "0" # 0 离线 1 在线
3) "online_at"
4) "1542272854" # 上线时间戳
5) "offline_at"
6) "undefined" # 离线时间戳

客户端代理订阅

配置项

打开配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf],配置 Backend 规则:
EMQ X 数据持久化 存储消息到 Redis_第6张图片

使用示例

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

## redis key 为 mqtt:sub:{client_id}
## HSET key {topic} {qos}
127.0.0.1:6379> HSET mqtt:sub:sub_client sub_client/upstream 1
(integer) 1

127.0.0.1:6379> HSET mqtt:sub:sub_client sub_client/downlink 1
(integer) 1

topic发布订阅

持久化发布消息

配置项

打开配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf],配置 Backend 规则,支持使用 topic 参数进行消息过滤,此处使用 # 通配符存储任意主题消息:
EMQ X 数据持久化 存储消息到 Redis_第7张图片

使用示例

在 EMQ X 管理控制台 WebSocket 页面中,使用 clientid sub_client 建立连接,向主题 upstream_topic 发布多条消息。针对每条消息, EMQ X 将持久化消息列表、消息详情两条记录。
EMQ X 数据持久化 存储消息到 Redis_第8张图片

常用命令

  • 查看消息列表命令:ZRANGE mqtt:msg:upstream_topic 0 -1

ZRANGE mqtt:msg:upstream_topic 0 -1

## 获取 upstream_topic 主题集合中所有 message id
127.0.0.1:6379> ZRANGE mqtt:msg:upstream_topic 0 -1
1) "Mjk2MTY1NzU0MDI1MTM4MDM4NTkwNzgzNTc2OTE4NTg5NDI"
2) "Mjk2MTY1NzUxNzExODcwNTgzODIyMzAzMDA0NTg0MTgxNzI"
3) "Mjk2MTY1NzUzMDcxOTgyNDEyOTgzNjgxNjU5NTIwMjg2NzF"
127.0.0.1:6379>
  • 查看消息详情命令:HGETALL mqtt:msg:{key}

EMQ X 数据持久化 存储消息到 Redis_第9张图片

## 获取 message id 为 Mjk2MTY1NzU0MDI1MTM4MDM4NTkwNzgzNTc2OTE4NTg5NDI 的消息详情
127.0.0.1:6379> HGETALL mqtt:msg:Mjk2MTY1NzU0MDI1MTM4MDM4NTkwNzgzNTc2OTE4NTg5NDI
 1) "id"
 2) "Mjk2MTY1NzU0MDI1MTM4MDM4NTkwNzgzNTc2OTE4NTg5NDI"
 3) "from"
 4) "sub_client"
 5) "qos"
 6) "1"
 7) "topic"
 8) "upstream_topic"
 9) "payload"
10) "{ \"msg\": \"\xe6\x88\x91\xe6\xad\xa3\xe5\x9c\xa8\xe5\xad\xa6\xe4\xb9\xa0EMQ-Redis\xe6\x8c\x81\xe4\xb9\x85\xe5\x8c\x96\xe5\x8f\x91\xe5\xb8\x83\xe6\xb6\x88\xe6\x81\xaf002\" }"
11) "ts"
12) "1605517769649"
13) "retain"
14) "false"

获取离线消息

配置项

打开配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf],配置 Backend 规则:
EMQ X 数据持久化 存储消息到 Redis_第10张图片

使用示例

MQTT 离线消息需满足以下条件:

1.   以 clean_session = false 连接
2.   订阅 QoS > 0
3.   发布 QoS > 0

在 EMQ X 管理控制台中新建一个客户端sub_client_another建立如下配置建立连接往topic:sub_client/upstream发布消息:
EMQ X 数据持久化 存储消息到 Redis_第11张图片

持久化 Retain 消息

打开配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf],配置 Backend 规则:
EMQ X 数据持久化 存储消息到 Redis_第12张图片

常用命令

  • 消息列表命令:ZRANGE mqtt:retain:upstream_topic 0 -1或者 GETRANGE mqtt:retain:upstream_topic 0 -1【根据redis存储的数据类型来获取value值】

EMQ X 数据持久化 存储消息到 Redis_第13张图片

  • 消息详情命令:HGETALL mqtt:msg:Mjk2MTY3MzA0ODEyNzgzNDMyNzI0NjM4ODU5NTQ4NDI2MjJ

EMQ X 数据持久化 存储消息到 Redis_第14张图片

## 获取 message id 为 Mjk2MTY3MzA0ODEyNzgzNDMyNzI0NjM4ODU5NTQ4NDI2MjJ 的消息详情
127.0.0.1:6379> HGETALL mqtt:msg:2VFt0kuNrOktefX6m4nP
 1) "id"
 2) "Mjk2MTY3MzA0ODEyNzgzNDMyNzI0NjM4ODU5NTQ4NDI2MjJ"
 3) "from"
 4) "sub_client"
 5) "qos"
 6) "2"
 7) "topic"
 8) "upstream_topic"
 9) "payload"
10) "{ \"msg\": \"now i am studing\" }"
11) "ts"
12) "1605526176485"
13) "retain"
14) "true"

至此EMQ X 数据持久化 存储消息到 Redis【场景一】模拟完毕!
【参考文档】EMQ官网链接:EMQ官网
下一级[场景二:将 EMQ X 指定主题下且满足条件的消息存储到 Reids]:将 EMQ X 指定主题下且满足条件的消息存储到 Reids

你可能感兴趣的