Cassandra 消息存储

提示

EMQX 3.1 版本后推出强大的规则引擎用于替换插件，建议您前往使用保存数据到 Cassandra规则引擎中创建 保存数据到 Cassandra

配置 Cassandra 服务器

配置文件: emqx_backend_cassa.conf

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

  1. ## Cassandra 节点地址
  2. backend.ecql.pool1.nodes = 127.0.0.1:9042
  4. ## Cassandra 连接池大小
  5. backend.ecql.pool1.size = 8
  7. ## Cassandra 自动重连间隔(s)
  8. backend.ecql.pool1.auto_reconnect = 1
  10. ## Cassandra 认证用户名/密码
  11. backend.ecql.pool1.username = cassandra
  12. backend.ecql.pool1.password = cassandra
  14. ## Cassandra Keyspace
  15. backend.ecql.pool1.keyspace = mqtt
  17. ## Cassandra Logger type
  18. backend.ecql.pool1.logger = info
  20. ##--------------------------------------------------------------------
  21. ## Cassandra Backend Hooks
  22. ##--------------------------------------------------------------------
  24. ## Client Connected Record
  25. backend.cassa.hook.client.connected.1    = {"action": {"function": "on_client_connected"}, "pool": "pool1"}
  27. ## Subscribe Lookup Record
  28. backend.cassa.hook.session.created.1     = {"action": {"function": "on_subscription_lookup"}, "pool": "pool1"}
  30. ## Client DisConnected Record
  31. backend.cassa.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}
  33. ## Lookup Unread Message QOS > 0
  34. backend.cassa.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch"}, "pool": "pool1"}
  36. ## Lookup Retain Message
  37. backend.cassa.hook.session.subscribed.2  = {"action": {"function": "on_retain_lookup"}, "pool": "pool1"}
  39. ## Store Publish Message  QOS > 0
  40. backend.cassa.hook.message.publish.1     = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}
  42. ## Delete Acked Record
  43. backend.cassa.hook.session.unsubscribed.1= {"topic": "#", "action": {"cql": ["delete from acked where clientid = ${clientid} and topic = ${topic}"]}, "pool": "pool1"}
  45. ## Store Retain Message
  46. backend.cassa.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "pool": "pool1"}
  48. ## Delete Retain Message
  49. backend.cassa.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}
  51. ## Store Ack
  52. backend.cassa.hook.message.acked.1       = {"topic": "#", "action": {"function": "on_message_acked"}, "pool": "pool1"}
  54. ## 获取离线消息
  55. ### "offline_opts": 获取离线消息的配置
  56. ####   - max_returned_count: 单次拉去的最大离线消息数目
  57. ####   - time_range: 仅拉去在当前时间范围的消息
  58. ## backend.cassa.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch"}, "offline_opts": {"max_returned_count": 500, "time_range": "2h"}, "pool": "pool1"}
  60. ## 如果需要存储 Qos0 消息, 可开启以下配置
  61. ## 提示: 当开启以下配置时, 需关闭 'on_message_fetch', 否则 qos1, qos2 消息会被存储俩次
  62. ## backend.cassa.hook.message.publish.4     = {"topic": "#", "action": {"function": "on_message_store"}, "pool": "pool1"}

backend 消息存储规则包括:

hooktopicaction说明
client.connectedon_client_connected存储客户端在线状态
session.createdon_subscribe_lookup订阅主题
client.disconnectedon_client_disconnected存储客户端离线状态
session.subscribed#on_message_fetch获取离线消息
session.subscribed#on_retain_lookup获取retain消息
session.unsubscribed#删除 akced 消息
message.publish#on_message_publish存储发布消息
message.publish#on_message_retain存储retain消息
message.publish#on_retain_delete删除retain消息
message.acked#on_message_acked消息ACK处理

自定义 CQL 语句 可用参数包括:

hook可用参数示例(cql语句中${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})

支持 CQL 语句配置:

考虑到用户的需求不同, backend cassandra 自带的函数无法满足用户需求, 用户可根据自己的需求配置 cql 语句

在 etc/plugins/emqx_backend_cassa.conf 中添加如下配置:

  1. ## 在客户端连接到 EMQX 服务器后，执行一条 cql 语句(支持多条 cql 语句)
  2. backend.cassa.hook.client.connected.3 = {"action": {"cql": ["insert into conn(clientid) values(${clientid})"]}, "pool": "pool1"}

Cassandra 创建一个 Keyspace

  1. CREATE KEYSPACE mqtt WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}  AND durable_writes = true;
  2. USE mqtt;

导入 Cassandra 表结构

  1. cqlsh -e "SOURCE 'emqx_backend_cassa.cql'"

Cassandra 用户状态表(Client Table)

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

  1. CREATE TABLE mqtt.client (
  2.   clientid text PRIMARY KEY,
  3.   connected timestamp,
  4.   disconnected timestamp,
  5.   node text,
  6.   state int
  7. );

查询设备在线状态:

  1. select * from mqtt.client where clientid = ${clientid};

例如 ClientId 为 test 的客户端上线:

  1. select * from mqtt.client where clientid = 'test';
  3.  clientid  | connected                       | disconnected  | node            | state
  4. -----------+---------------------------------+---------------+-----------------+-------
  5.       test | 2017-02-14 08:27:29.872000+0000 |          null | emqx@127.0.0.1|   1

例如ClientId为test客户端下线:

  1. select * from mqtt.client where clientid = 'test';
  3.  clientid  | connected                       | disconnected                    | node            | state
  4. -----------+---------------------------------+---------------------------------+-----------------+-------
  5.       test | 2017-02-14 08:27:29.872000+0000 | 2017-02-14 08:27:35.872000+0000 | emqx@127.0.0.1|     0

Cassandra 用户订阅主题表(Sub Table)

mqtt.sub 存储订阅关系:

  1. CREATE TABLE mqtt.sub (
  2.   clientid text,
  3.   topic text,
  4.   qos int,
  5.   PRIMARY KEY (clientid, topic)
  6. );

用户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';
  3.  clientid  | topic       | qos
  4. -----------+-------------+------
  5.       test | test_topic1 |   1
  6.       test | test_topic2 |   2

Cassandra 发布消息表(Msg Table)

mqtt.msg 存储MQTT消息:

  1. CREATE TABLE mqtt.msg (
  2.   topic text,
  3.   msgid text,
  4.   arrived timestamp,
  5.   payload text,
  6.   qos int,
  7.   retain int,
  8.   sender text,
  9.   PRIMARY KEY (topic, msgid)
  10. ) WITH CLUSTERING ORDER BY (msgid DESC);

查询某个客户端发布的消息:

  1. select * from mqtt_msg where sender = ${clientid};

查询ClientId为’test’的客户端发布的消息:

  1. select * from mqtt_msg where sender = 'test';
  3.  topic | msgid                | arrived                         | payload      | qos | retain | sender
  4. -------+----------------------+---------------------------------+--------------+-----+--------+--------
  5.  hello | 2PguFrHsrzEvIIBdctmb | 2017-02-14 09:07:13.785000+0000 | Hello world! |   1 |      0 |   test
  6.  world | 2PguFrHsrzEvIIBdctmb | 2017-02-14 09:07:13.785000+0000 | Hello world! |   1 |      0 |   test

Cassandra 保留消息表(Retain Message Table)

mqtt.retain 存储 Retain 消息:

  1. CREATE TABLE mqtt.retain (
  2.   topic text PRIMARY KEY,
  3.   msgid text
  4. );

查询 retain 消息:

  1. select * from mqtt_retain where topic = ${topic};

查询 topic 为 ‘t/retain’ 的 retain 消息:

  1. select * from mqtt_retain where topic = 't/retain';
  3.  topic  | msgid
  4. --------+----------------------
  5.  retain | 2PguFrHsrzEvIIBdctmb

Cassandra 接收消息 ack 表(Message Acked Table)

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

  1. CREATE TABLE mqtt.acked (
  2.   clientid text,
  3.   topic text,
  4.   msgid text,
  5.   PRIMARY KEY (clientid, topic)
  6. );

启用 Cassandra 存储插件

  1. ./bin/emqx_ctl plugins load emqx_backend_cassa