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

配置 Redis 存储规则

backend.redis.hook.session.created.1     = {"action": {"function": "on_subscribe_lookup"}, "pool": "pool1"}
backend.redis.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}
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"}
backend.redis.hook.session.subscribed.3  = {"action": {"function": "on_retain_lookup"}, "pool": "pool1"}
backend.redis.hook.session.unsubscribed.1= {"topic": "#", "action": {"commands": ["DEL mqtt:acked:${clientid}:${topic}"]}, "pool": "pool1"}
backend.redis.hook.message.publish.1     = {"topic": "#", "action": {"function": "on_message_publish"}, "expired_time" : 3600, "pool": "pool1"}
backend.redis.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "expired_time" : 3600, "pool": "pool1"}
backend.redis.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}
backend.redis.hook.message.acked.1       = {"topic": "queue/#", "action": {"function": "on_message_acked_for_queue"}, "pool": "pool1"}
backend.redis.hook.message.acked.2       = {"topic": "pubsub/#", "action": {"function": "on_message_acked_for_pubsub"}, "pool": "pool1"}
## backend.redis.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch_for_keep_latest"}, "pool": "pool1"}
## backend.redis.hook.message.publish.1     = {"topic": "#", "action": {"function": "on_message_store_keep_latest"}, "expired_time" : 3600, "pool": "pool1"}
## backend.redis.hook.message.acked.1       = {"topic": "#", "action": {"function": "on_message_acked_for_keep_latest"}, "pool": "pool1"}

Redis 存储规则说明

Redis 命令行参数说明

Redis 存储支持用户采用 Redis Commands 语句配置 Action，例如:

## 在客户端连接到 EMQ X 服务器后，执行一条 redis

Redis 设备在线状态 Hash

mqtt:client Hash 存储设备在线状态:

hmset
key = mqtt:client:${clientid}
value = {state:int, online_at:timestamp, offline_at:timestamp}
hset
key = mqtt:node:${node}
field = ${clientid}
value = ${ts}

查询设备在线状态:

例如 ClientId 为 test 客户端上线:

HGETALL mqtt:client:test
1) "state"
2) "1"
3) "online_at"
4) "1481685802"
5) "offline_at"
6) "undefined"

例如 ClientId 为 test 客户端下线:

Redis 保留消息 Hash

hmset
key = mqtt:retain:${topic}
value = {id: string, from: string, qos: int, topic: string, retain: int, payload: string, ts: timestamp}

查询 retain 消息:

HGETALL "mqtt:retain:${topic}"

例如查看 topic 为 topic 的 retain 消息:

HGETALL mqtt:retain:topic
 1) "id"
    2) "6P9NLcJ65VXBbC22sYb4"
 3) "from"
    4) "test"
 5) "qos"
 6) "1"
 7) "topic"
 8) "topic"
10) "true"
11) "payload"
12) "Hello world!"
13) "ts"

Redis 消息存储 Hash

mqtt:msg Hash 存储 MQTT 消息:

hmset
key = mqtt:msg:${msgid}
value = {id: string, from: string, qos: int, topic: string, retain: int, payload: string, ts: timestamp}
zadd
key = mqtt:msg:${topic}
field = 1
value = ${msgid}

mqtt:acked SET 存储客户端消息确认:

set
key = mqtt:acked:${clientid}:${topic}
value = ${msgid}

Redis 订阅存储 Hash

mqtt:sub Hash 存储订阅关系:

某个客户端订阅主题:

HSET mqtt:sub:${clientid} ${topic} ${qos}

HSET "mqtt:sub:test" "topic1" 1
HSET "mqtt:sub:test" "topic2" 2

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

HGETALL mqtt:sub:test
1) "topic1"
2) "1"
3) "topic2"
4) "2"

Redis SUB/UNSUB 事件发布

设备需要订阅/取消订阅主题时，业务服务器向 Redis 发布事件消息:

PUBLISH
channel = "mqtt_channel"
message = {type: string , topic: string, clientid: string, qos: int}
\*type: [subscribe/unsubscribe]

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

PUBLISH "mqtt_channel" "{\"type\": \"subscribe\", \"topic\": \"topic0\", \"clientid\": \"test\", \"qos\": \"0\"}"

例如 ClientId 为 test 客户端取消订阅主题:

Tip

Redis Cluster 无法使用 Redis PUB/SUB 功能。

启用 Redis 数据存储插件