EMQ百万级MQTT消息服务

EMQ官方地址:http://emqtt.com/

EMQ中文文档:http://emqtt.com/docs/v2/guide.html

1.ACL鉴权规则化

在正常业务使用下对于客户端的行为可以使用ACL进行限制,比如A客户端只能订阅 /A/get 队列消息和向 /A/set 发布内容 但是在MYSQL里面处理这样的鉴权就需要写入两条记录,如果设备量有一百万数据库就要承担两百万条鉴权数据量会大大影响数据库的性能 那么有没有什么批量的方式来定义ACL鉴权呢?

在mysql-ACL鉴权的配置文件下关于如何使用鉴权的SQL是可以编辑的,也就意味着你可以通过SQL来实现批量ACL鉴权规则

> vim /usr/local/emqttd/etc/plugins/emq_auth_mysql.conf# 最下面有这样一条配置auth.mysql.acl_query = select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'

笔者这里就实现每个设备默认可以订阅 /A/get 队列消息和向 /A/set 发布

笔者现在的规则是客户端只能向hello写消息其他操作一概不允许,我们先加两条记录

insert `mqtt_acl`(`allow`,`username`,`access`,`topic`) values(1,'$all',1,'/$user/get');insert `mqtt_acl`(`allow`,`username`,`access`,`topic`) values(1,'$all',2,'/$user/set');

然后修改默认的ACL鉴权的SQL语句如下(这里使用的是username作为topic动态名称也可以使用其他字段):

select allow, ipaddr, username, clientid, access, REPLACE(topic,'$user','%u') from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c'

这样一来就算没有独立配置/A/set可以写入,作为用户是A的客户端也可以进行消息的写入了,并且也可以监听消息/A/get

2.共享订阅

关于队列常见的使用中也有这样的场景,一条消息希望被多个监听程序接收到,可能的场景如下:

  • 一个程序处理,一个程序记录日志分别处理
  • 批量推送
                            ---------
                            |       | --Msg1,Msg2,Msg3--> Subscriber1Publisher--Msg1,Msg2,Msg3-->|  EMQ  | --Msg1,Msg2,Msg3--> Subscriber2
                            |       | --Msg1,Msg2,Msg3--> Subscriber3
                            ---------

多条消息希望被多个程序中的某个进行处理,场景如下:

  • 并发情况下耗时操作进行并行处理提高系统吞吐量
                            ---------
                            |       | --Msg1--> Subscriber1Publisher--Msg1,Msg2,Msg3-->|  EMQ  | --Msg2--> Subscriber2
                            |       | --Msg3--> Subscriber3
                            ---------

在默认情况下有多个客户端监听一个事件时会受到同样的消息,但是怎么共享订阅呢?EMQ共享订阅支持两种使用方式:

  • $queue/ 如:$queue/topic
  • $share/<group>/ 如:$share/group/topic

以上两种都可以实现共享订阅(笔者测试下来值通过了share来完成了订阅),订阅和监听

  • 多个服务端监听 $share/group/topic
  • 客户端向 topic 发送消息

3.Qos 0/1/2的区别实测

最多一次的传输 消息是基于TCP/IP网络传输的。没有回应,在协议中也没有定义重传的语义。消息可能到达服务器1次,也可能根本不会到达。

至少一次的传输 服务器接收到消息会被确认,通过传输一个PUBACK信息。如果有一个可以辨认的传输失败,无论是通讯连接还是发送设备,还是过了一段时间确认信息没有收到,发送方都会将消息头的DUP位置1,然后再次发送消息。消息最少一次到达服务器。SUBSCRIBE和UNSUBSCRIBE都使用level 1 的QoS。 如果客户端没有接收到PUBACK信息(无论是应用定义的超时,还是检测到失败然后通讯session重启),客户端都会再次发送PUBLISH信息,并且将DUP位置1。 当它从客户端接收到重复的数据,服务器重新发送消息给订阅者,并且发送另一个PUBACK消息。

笔者做了一个实现消费端阻塞2秒消费一个内容,发布端1秒发布一个内容,等EMQ的最大拥塞使用完了之后消息在EMQ缓存的会后就会出现很多的重复消息

只有一次的传输 在QoS level 1上附加的协议流保证了重复的消息不会传送到接收的应用。这是最高级别的传输,当重复的消息不被允许的情况下使用。这样增加了网络流量,但是它通常是可以接受的,因为消息内容很重要。 QoS level 2在消息头有Message ID。

4.密码加盐

在用户验证中可以使用plain | md5 | sha | sha256 | bcrypt等hash方式(默认使用的sha256),但是出于安全性考虑EMQ也支持对密码加盐,可以解开注释使用一下加盐方式中的一种

vim /usr/local/emqttd/etc/plugins/emq_auth_mysql.conf## sha256 with salt prefix## auth.mysql.password_hash = salt,sha256## bcrypt with salt only prefix## auth.mysql.password_hash = salt,bcrypt## sha256 with salt suffix## auth.mysql.password_hash = sha256,salt## pbkdf2 with macfun iterations dklen## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512## auth.mysql.password_hash = pbkdf2,sha256,1000,20

对应存储的密码就要进行加盐处理了

5.EMQ离线消息

  • 保留消息 MQTT客户端向服务器发布(PUBLISH)消息时,可以设置保留消息(Retained Message)标志。保留消息(Retained Message)会驻留在消息服务器,后来的订阅者订阅主题时仍可以接收该消息。 例如mosquitto命令行发布一条保留消息到主题’a/b/c’: mosquitto_pub -r -q 1 -t a/b/c -m 'hello' 之后连接上来的MQTT客户端订阅主题’a/b/c’时候,仍可收到该消息: $ mosquitto_sub -t a/b/c -q 1 hello 保留消息(Retained Message)有两种清除方式: 客户端向有保留消息的主题发布一个空消息: mosquitto_pub -r -q 1 -t a/b/c -m '' 消息服务器设置保留消息的超期时间。
  • cleanSession 清理回话 MQTT客户端向服务器发起CONNECT请求时,可以通过’Clean Session’标志设置会话。 ‘Clean Session’设置为0,表示创建一个持久会话,在客户端断开连接时,会话仍然保持并保存离线消息,直到会话超时注销。 ‘Clean Session’设置为1,表示创建一个新的临时会话,在客户端断开时,会话自动销毁。

3 总结

在EMQ和MQTT使用过程中还有很多的细节需要注意,关注细节才能走的更远

希望本文对你有帮助,求帮转,谢谢

原文发布于微信公众号 - IT技术精选文摘(ITHK01)

原文发表时间:2018-04-22

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏青蛙要fly的专栏

Android技能树 — 网络小结(4)之socket/websocket/webservice

介于自己的网络方面知识烂的一塌糊涂,所以准备写相关网络的文章,但是考虑全部写在一篇太长了,所以分开写,希望大家能仔细看,最好可以指出我的错误,让我也能纠正。

693
来自专栏向治洪

jsoup详解

json相信大家都用的多,jsonp我就一直没有机会用到,但也经常看到,只知道是“用来跨域的”,一直不知道具体是个什么东西。今天总算搞明白了。下面一步步来搞清...

1909
来自专栏喵了个咪的博客空间

原 荐 EMQ百万级MQTT消息服务(小技巧)

附上: 喵了个咪的博客:w-blog.cn EMQ官方地址:http://emqtt.com/ EMQ中文文档:http://emqtt.com/docs/v2...

3364
来自专栏ionic3+

【技巧】ionic3自动聚焦暴力实现

很早前和群里的人探讨过自动聚焦,在android上可以,但是在ios上失败,后来在网上看到这个:

602
来自专栏友弟技术工作室

Python 资源大全中文版

GitHub 上有一个 Awesome - XXX 系列的资源整理,资源非常丰富,涉及面非常广。awesome-python 是 vinta 发起维护的 Pyt...

3495
来自专栏静默虚空的博客

Webpack 资源管理

资源管理 ? 提示: 1、版本问题 本文基于 webpack 2.x 版本。webpack 2.x 相比 webpack 1.x 有重大改变。所以,如果你的...

1967
来自专栏张善友的专栏

微信官方开源UI库-WeUI

概述   WeUI是一套同微信原生视觉体验一致的基础样式库,为微信Web开发量身设计,可以令用户的使用感知更加统一。包含button、cell、dialog、...

2645
来自专栏一个会写诗的程序员的博客

React Router 之 browserHistoryHistoriesHistories

前端工程采用 SPA 模式 hashHistory , 在集成到生产环境中的时候,使用browserHistory :

732
来自专栏phodal

听我说说我的博客: 月访问量过万的个人IT博客的技术史

我的博客是如何工作的? HTTP服务器 当你开发在网页上访问我的博客的时候,你可能会注意到上面的协议是HTTPS。 ? 但是并不会察觉到它是HTTP2.0。而这...

20510
来自专栏为数不多的Android技巧

简化markdown写作中的贴图流程

这么复杂的流程,让人简直没有了插入图片的欲望;但是大量的文字没有图片,必然让人疲惫;

1015

扫码关注云+社区