离线消息保存到 Cassandra

搭建 Cassandra 数据库,并设置用户名密码为 root/public,以 MacOS X 为例:

  1. $ brew install cassandra
  2. ## 修改配置,关闭匿名认证
  3. $ vim /usr/local/etc/cassandra/cassandra.yaml
  4. authenticator: PasswordAuthenticator
  5. authorizer: CassandraAuthorizer
  6. $ brew services start cassandra
  7. ## 创建 root 用户
  8. $ cqlsh -ucassandra -pcassandra
  9. create user root with password 'public' superuser;

初始化 Cassandra 表空间:

  1. $ cqlsh -uroot -ppublic
  2. CREATE KEYSPACE mqtt WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'} AND durable_writes = true;

创建 mqtt_msg 表:

  1. CREATE TABLE mqtt.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)
  11. AND bloom_filter_fp_chance = 0.01
  12. AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
  13. AND comment = ''
  14. AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
  15. AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
  16. AND crc_check_chance = 1.0
  17. AND dclocal_read_repair_chance = 0.1
  18. AND default_time_to_live = 0
  19. AND gc_grace_seconds = 864000
  20. AND max_index_interval = 2048
  21. AND memtable_flush_period_in_ms = 0
  22. AND min_index_interval = 128
  23. AND read_repair_chance = 0.0
  24. AND speculative_retry = '99PERCENTILE';

消息表结构不能修改,请使用上面SQL语句创建

创建规则:

打开 EMQ X Dashboard,选择左侧的“规则”选项卡。

然后填写规则 SQL:

FROM说明

t/#: 发布者发布消息触发保存离线消息到Cassandra

$events/session_subscribed: 订阅者订阅主题触发获取离线消息

$events/message_acked: 订阅者回复消息ACK后触发删除已经被接收的离线消息

  1. SELECT * FROM "t/#", "$events/session_subscribed", "$events/message_acked" WHERE topic =~ 't/#'

离线消息保存到 Cassandra - 图1

关联动作:

在“响应动作”界面选择“添加动作”,然后在“动作”下拉框里选择“离线消息保存到 Cassandra ”。

离线消息保存到 Cassandra - 图2

现在资源下拉框为空,可以点击右上角的 “新建” 来创建一个 Cassandra 资源:

离线消息保存到 Cassandra - 图3

弹出一个“创建资源”对话框

离线消息保存到 Cassandra - 图4

填写资源配置:

填写真实的 Cassandra 服务器地址,其他配置填写相应的值,然后点击 “测试连接” 按钮,确保连接测试成功。

最后点击 “确定” 按钮。

离线消息保存到 Cassandra - 图5

返回响应动作界面,点击 “确认”。

离线消息保存到 Cassandra - 图6

返回规则创建界面,点击 “创建”。

离线消息保存到 Cassandra - 图7

规则已经创建完成,通过 Dashboard 的 WebSocket 客户端发一条数据(发布消息的QoS必须大于0):

离线消息保存到 Cassandra - 图8

消息发送后,通过 cqlsh 查看到消息被保存到 Cassandra 里面:

离线消息保存到 Cassandra - 图9

使用另外一个客户端,订阅主题 “t/1” (订阅主题的QoS必须大于0,否则消息会被重复接收):

离线消息保存到 Cassandra - 图10

订阅后马上接收到了保存到 Cassandra 里面的离线消息:

离线消息保存到 Cassandra - 图11

离线消息被接收后会在 Cassandra 中删除:

离线消息保存到 Cassandra - 图12