EMQ X 插件持久化系列 （三）EMQ X Redis 數據持久化

EMQ X 數據持久化簡介

數據持久化的主要使用場景包括將客戶端上下線狀態，訂閱主題信息，消息內容，消息抵達後發送消息回執等操做記錄到 Redis、MySQL、PostgreSQL、MongoDB、Cassandra、AWS DynamoDB 等各種數據庫中供外部服務快速查詢或在服務宕機/客戶端異常離線時保留當前運行狀態，鏈接恢復時恢復到以前狀態；持久化亦可用於客戶端代理訂閱，設備客戶端上線時，持久化模塊直接從數據庫加載預設的主題並完成代理訂閱，下降系統設計複雜度和減小客戶端訂閱通訊開銷。

用戶也能夠經過訂閱相關主題的方式來實現相似的功能，可是在企業版中內置的這些持久化的支持執行效率更高、可靠性更強，大大下降了開發者的工做量並提高了系統穩定性。

數據持久化是 EMQ X 的重要功能，僅在企業版支持。

持久化設計

持久化原理是配置事件鉤子觸發時調用處理函數（action）,處理函數獲取到相應的數據後按照配置的指令進行處理，實現數據的增、刪、改、查。相同事件鉤子在不一樣數據庫中可用參數是同樣的，但處理函數（action）因數據庫特性不一樣有所差別。整個持久化工做模式和流程以下：

一對一消息存儲

1. Publish 端發佈一條消息；
2. Backend 將消息記錄數據庫中；
3. Subscribe 端訂閱主題；
4. Backend 從數據庫中獲取該主題的消息；
5. 發送消息給 Subscribe 端；
6. Subscribe 端確認後 Backend 從數據庫中移除該消息；

一對多消息存儲

1. PUB 端發佈一條消息；
2. Backend 將消息記錄在數據庫中；
3. SUB1 和 SUB2 訂閱主題；
4. Backend 從數據庫中獲取該主題的消息；
5. 發送消息給 SUB1 和 SUB2；
6. Backend 記錄 SUB1 和 SUB2 已讀消息位置，下次獲取消息從該位置開始。

Redis 數據持久化

本文以實際例子來講明如何經過 Redis 來存儲相關的信息。

Redis 是徹底開源免費遵照 BSD 協議的高性能 key-value 數據庫。

相比其餘 key-value 緩存產品 Redis 有如下特色：

• Redis 性能極高，單機支持十萬級別的讀寫速度。
• Redis 支持數據的持久化，能夠將內存中的數據保存在磁盤中，重啓的時候能夠再次加載進行使用。
• Redis 不只僅支持簡單的 key-value 類型的數據，同時還提供 list，set，zset，hash 等數據結構的存儲。
• Redis 支持數據的備份，即 master-slave 模式的數據備份。

讀者能夠參考 Redis 官方的 Quick Start 來安裝 Redis（寫本文的時候，Redis 版本爲5.0），經過 redis-server 命令來啓動 Redis 服務器。

配置 EMQ X 服務器

經過 RPM 方式安裝的 EMQ X，Redis 相關的配置文件位於 /etc/emqx/plugins/emqx_backend_redis.conf，若是隻是測試 Redis 持久化的功能，大部分配置不須要作更改。惟一須要更改的地方多是 Redis 服務器的地址：若是讀者安裝的 Redis 不與 EMQ X 在同一服務器上，請指定正確的 Redis 服務器的地址與端口。以下所示：

## Redis Server 127.0.0.1:6379, Redis Sentinel: 127.0.0.1:26379
backend.redis.pool1.server = 127.0.0.1:6379

保持剩下部分的配置文件不變，而後啓動該插件：

emqx_ctl plugins load emqx_backend_redis

客戶端在線狀態存儲

客戶端上下線時，更新在線狀態、上下線時間、節點客戶端列表至 Redis 數據庫。

儘管 EMQ X 自己提供了設備在線狀態 API，但在須要頻繁獲取客戶端在線狀態、上下線時間的場景下，直接從數據庫獲取該記錄比調用 EMQ X API 更高效。

配置項

打開配置文件，配置 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://127.0.0.1:18083 EMQ X 管理控制檯，在 工具 -> Websocket 中新建一個客戶端鏈接，指定 clientid 爲 sub_client:

打開 redis-cli 命令行窗口，執行命令 keys *，結果以下所示，讀者能夠看到在 Redis 中存儲了兩個 key：

127.0.0.1:6379> keys *
1) "mqtt:node:emqx@127.0.0.1"
2) "mqtt:client:sub_client"

鏈接列表

插件以 mqtt:node:{node_name} 格式的 key 記錄節點下客戶端列表及鏈接時間戳信息，等效操做：

## redis key 爲 mqtt:node:{node_name}
HMSET mqtt:node:emqx@127.0.0.1 sub_client 1542272836

字段說明：

## 節點下在線設備信息
127.0.0.1:6379> HGETALL mqtt:node:emqx@127.0.0.1
1) "sub_client1" # clientid
2) "1542272836" # 上線時間時間戳
3) "sub_client"
4) "1542272836"

鏈接詳細信息

插件以 mqtt:client:{client_id} 格式的 key 記錄客戶端在線狀態、上線時間，等效操做：

## redis key 爲 mqtt:client:{client_id}
HMSET mqtt:client:sub_client state 1 online_at 1542272854

字段說明：

## 客戶端在線狀態
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" # 離線時間戳

客戶端代理訂閱

客戶端上線時，存儲模塊直接從數據庫讀取預設待訂閱列表，代理加載訂閱主題。在客戶端須要經過預約主題通訊（接收消息）場景下，應用能從數據層面設定 / 改變代理訂閱列表。

配置項

打開配置文件，配置 Backend 規則：

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

使用示例

當 sub_client 設備上線時，須要爲其訂閱 sub_client/upstream 與 sub_client/downlink 兩個 QoS 1 的主題：

1. 插件以 mqtt:sub:{client_id} 格式 key 在 Redis 中初始化代理訂閱 Hash：

## 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) 0

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

2. EMQ X 管理控制檯 WebSocket 頁面，以 clientid sub_client 新建一個客戶端鏈接，切換至訂閱頁面，可見當前客戶端自動訂閱了 sub_client/upstream 與 sub_client/downlink 兩個 QoS 1 的主題：

3. 切換回管理控制檯 WebSocket 頁面，向 sub_client/downlink 主題發佈消息，可在消息訂閱列表收到發佈的消息。

持久化發佈消息

配置項

打開配置文件，配置 Backend 規則，支持使用 topic 參數進行消息過濾，此處使用 # 通配符存儲任意主題消息：

## hook: message.publish
## action/function: on_message_publish

backend.redis.hook.message.publish.1 = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}

使用示例

在 EMQ X 管理控制檯 WebSocket 頁面中，使用 clientid sub_client 創建鏈接，向主題 upstream_topic 發佈多條消息。針對每條消息， EMQ X 將持久化消息列表、消息詳情兩條記錄。

消息列表

EMQ X 將消息列表以 message id 持久化至 mqtt:msg:{topic} Redis 集合中：

## 獲取 upstream_topic 主題集合中全部 message id
127.0.0.1:6379> ZRANGE mqtt:msg:upstream_topic 0 -1
1) "2VFsyhDm0cPIQvnY9osj"
2) "2VFszTClyjpVtLDLrn1u"
3) "2VFszozkwkYOcbEy8QN9"
4) "2VFszpEc7DfbEqC97I3g"
5) "2VFszpSzRviADmcOeuXd"
6) "2VFszpm3kvvLkJTcdmGU"
7) "2VFt0kuNrOktefX6m4nP"
127.0.0.1:6379>

消息詳情

每條消息詳情將以 mqtt:msg:{message_id} 格式的 key 存儲在 Redis Hash 中：

## 獲取 message id 爲 2VFt0kuNrOktefX6m4nP 的消息詳情
127.0.0.1:6379> HGETALL mqtt:msg:2VFt0kuNrOktefX6m4nP
 1) "id"
 2) "2VFt0kuNrOktefX6m4nP" ## message id
 3) "from"
 4) "sub_client" ## client id
 5) "qos"
 6) "2"
 7) "topic"
 8) "up/upstream_topic"
 9) "payload"
10) "{ \"cmd\": \"reboot\" }"
11) "ts"
12) "1542338754" ## pub 時間戳
13) "retain"
14) "false"

獲取離線消息

配置項

打開配置文件，配置 Backend 規則：

## hook: session.subscribed
## action/function: on_message_fetch_for_queue、on_message_fetch_for_pubsub

## 一對一離線消息
backend.redis.hook.session.subscribed.1  = {"topic": "queue/#", "action": {"function": "on_message_fetch_for_queue"}, "pool": "pool1"}

## 一對多離線消息
backend.redis.hook.session.subscribed.2  = {"topic": "pubsub/#", "action": {"function": "on_message_fetch_for_pubsub"}, "pool": "pool1"}

使用示例

MQTT 離線消息需知足如下條件：

1. 以 clean_session = false 鏈接
2. 訂閱 QoS > 0
3. 發佈 QoS > 0

在 EMQ X 管理控制檯中以以下配置創建鏈接，

持久化 Retain 消息

配置項

打開配置文件，配置 Backend 規則：

## hook: message.publish
## action/function: on_client_connected、on_message_retain

backend.redis.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "pool": "pool1"}

backend.redis.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}

消息列表

EMQ X 將消息列表以 message id 持久化至 mqtt:retain:{topic} Redis Hash 中：

## 獲取 upstream_topic 主題集合中全部 message id
127.0.0.1:6379> ZRANGE mqtt:retain:upstream_topic 0 -1
1) "2VFsyhDm0cPIQvnY9osj"
127.0.0.1:6379>

消息詳情

每條消息詳情將以 mqtt:msg:{message_id} 格式的 key 存儲在 Redis Hash 中：

## 獲取 message id 爲 2VFt0kuNrOktefX6m4nP 的消息詳情
127.0.0.1:6379> HGETALL mqtt:msg:2VFt0kuNrOktefX6m4nP
 1) "id"
 2) "2VFt0kuNrOktefX6m4nP" ## message id
 3) "from"
 4) "sub_client" ## client id
 5) "qos"
 6) "2"
 7) "topic"
 8) "up/upstream_topic"
 9) "payload"
10) "{ \"cmd\": \"reboot\" }"
11) "ts"
12) "1542338754" ## pub 時間戳
13) "retain"
14) "false"

總結

讀者在理解了 Redis 中所存儲的數據結構以後，能夠利用各類 Redis 客戶端來實現對相關信息的讀取，

---

更多信息請訪問咱們的官網 emqx.io，或關注咱們的開源項目 github.com/emqx/emqx ，詳細文檔請訪問 官方文檔。