关于java: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如下所示：

常用命令

查看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

启动插件

命令：emqx_ctl plugins load emqx_backend_redis

或者到EMQ X Dashboard 治理控制台，在插件 -> 全副-> emqx_backend_redis 中启动插件

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

客户端高低线时，更新在线状态、高低线工夫、节点客户端列表至 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"}

应用示例

浏览器关上 http://ip:18083 EMQ X Dashboard 治理控制台，在工具 -> Websocket 中新建一个客户端连贯，指定 clientid 为 sub_client

执行命令：redis-cli -h 127.0.0.1 -p 6379或者/usr/local/soft/redis5/bin/redis-cli 关上 redis-cli 命令行窗口，执行命令 keys *，后果如下所示，读者能够看到在 Redis 中存储了两个 key：

如果redis设置了明码,应用命令：auth passward受权

常用命令

连贯列表命令：HGETALL mqtt:node:emqx@127.0.0.1

字段阐明：

## 节点下在线设施信息127.0.0.1:6379> HGETALL mqtt:node:emqx@127.0.0.11) "sub_client1" # clientid2) "1542272836" # 上线工夫工夫戳

连贯详细信息命令：HGETALL mqtt:client:sub_client

字段阐明：

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

客户端代理订阅

配置项

关上配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf]，配置 Backend 规定：

应用示例

当 sub_client 设施上线时，须要为其订阅 sub_client/upstream 与 sub_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) 1127.0.0.1:6379> HSET mqtt:sub:sub_client sub_client/downlink 1(integer) 1

回到Dashboard中，查看订阅列表，可见以后客户端主动订阅了 sub_client/upstream 与 sub_client/downlink 两个 QoS 1 的主题：

切换回治理控制台 WebSocket 页面，向 sub_client/downlink 主题公布音讯，可在音讯订阅列表收到公布的音讯。

长久化公布音讯

配置项

关上配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf]，配置 Backend 规定，反对应用 topic 参数进行音讯过滤，此处应用 # 通配符存储任意主题音讯：

应用示例

在 EMQ X 治理控制台 WebSocket 页面中，应用 clientid sub_client 建设连贯，向主题 upstream_topic 公布多条音讯。针对每条音讯， EMQ X 将长久化音讯列表、音讯详情两条记录。

常用命令

查看音讯列表命令：ZRANGE mqtt:msg:upstream_topic 0 -1

## 获取 upstream_topic 主题汇合中所有 message id127.0.0.1:6379> ZRANGE mqtt:msg:upstream_topic 0 -11) "Mjk2MTY1NzU0MDI1MTM4MDM4NTkwNzgzNTc2OTE4NTg5NDI"2) "Mjk2MTY1NzUxNzExODcwNTgzODIyMzAzMDA0NTg0MTgxNzI"3) "Mjk2MTY1NzUzMDcxOTgyNDEyOTgzNjgxNjU5NTIwMjg2NzF"127.0.0.1:6379>

查看音讯详情命令：HGETALL mqtt:msg:{key}

## 获取 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 规定:

应用示例

MQTT 离线音讯需满足以下条件：

1.   以 clean_session = false 连贯2.   订阅 QoS > 03.   公布 QoS > 0

在 EMQ X 治理控制台中新建一个客户端sub_client_another建设如下配置建设连贯往topic：sub_client/upstream公布音讯：

再断开sub_client_another的连贯，从新上线sub_client,能够看到仍能收到音讯：

长久化 Retain 音讯

关上配置文件[配置文件目录/etc/emqx/plugins/emqx_backend_redis.conf]，配置 Backend 规定:

在Dashboard中公布Retain音讯：

常用命令

音讯列表命令：ZRANGE mqtt:retain:upstream_topic 0 -1或者 GETRANGE mqtt:retain:upstream_topic 0 -1【依据redis存储的数据类型来获取value值】

音讯详情命令：HGETALL mqtt:msg:Mjk2MTY3MzA0ODEyNzgzNDMyNzI0NjM4ODU5NTQ4NDI2MjJ

## 获取 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