阿里云微消息队列MQTT版（消息队列MQTT）

本文目录一览：

• 1、阿里云微消息队列（MQTT）的基本使用
• 2、阿里云mqtt收费标准
• 3、MQTT 消息队列遥感传输协议
• 4、在云服务器上搭建了mqtt，为什么手机连接不上mqtt，要怎么做才能连接上？求求大神帮忙
• 5、阿里云消息中间件（MQ）探秘
• 6、NE35E MQTT协议对接阿里云

阿里云微消息队列（MQTT）的基本使用

最近应系统功能需求，采购了一款云喇叭的物联网设备，就是插着4G卡那种，可以播放各种语音，仔细阅读了开发文档之后发现使用的是MQTT的协议，记录一下在对接中遇到的各种问题

MQTT是一个轻量的发布订阅模式消息传输协议，专门针对低带宽和不稳定网络环境的物联网应用设计。

MQTT特点：

阿里云的MQTT有两个版本，这里只说没有RocketMQ依赖的3.1.1及以上版本。

这里会自动生成用户名密码

阿里云mqtt收费标准

以下是阿里云mqtt收费标准：

其中铂金版实例、标准版实例、轻量版实例是包年包月（预付费）。是一种预付费模式，即先付费再使用。一般适用于业务量较大且长期运行的场景，通过包年包月付费方式您可以提前预留资源，同时享受更大的价格优化，帮您最大程度节省成本。

还有就是按量付费实例是按量付费（后付费）。即是一种后付费模式，即先使用再付费。一般适用于业务流量波峰波谷差异明显或临时测试的场景，可以有效避免资源浪费。

微消息队列mqtt版费用组成：

按量付费包括：同时在线连接数；消息收发量；订阅关系数

包年包月包括：连接数上限；消息TPS上限；订阅关系数上限

计价倍率介绍：

MQTT协议QoS=0且cleanSession=true 1

MQTT协议QoS=0且cleanSession=false 1

MQTT协议QoS=1且cleanSession=true 2

MQTT协议QoS=1且cleanSession=false 5

MQTT协议QoS=2且cleanSession=true 5

MQTT 消息队列遥感传输协议

1、 什么是MQTT?

MQTT（MessageQueueing Telemetry Transport Protocol）的全称是消息队列遥感传输协议的缩写，是由IBM公司推出的一种基于轻量级代理的发布/订阅模式的消息传输协议，运行在TCP协议栈之上，为其提供有序、可靠、双向连接的网络连接保证。由于其开放、简单和易于实现所以能够应用在资源受限的环境中，对于M2M和物联网应用程序来说是一个相当不错的选择。

2、 为什么要用MQTT？

MQTT协议是针对如下情况设计的：

M2M（Machine to Machine） communication，机器端到端通信，比如传感器之间的数据通讯 因为是Machine to Machine，需要考虑： Machine，或者叫设备，比如温度传感器，硬件能力很弱，协议要考虑尽量小的资源消耗，比如计算能力和存储等 M2M可能是无线连接，网络不稳定，带宽也比较小

MQTT的特点:

发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。这一点很类似于1. 这里是列表文本XMPP，但是MQTT的信息冗余远小于XMPP.

对负载内容屏蔽的消息传输。

使用TCP/IP提供网络连接。主流的MQTT是基于TCP连接进行数据推送的，但是同样有基于UDP的版本，叫做MQTT-SN。这两种版本由于基于不同的连接方式，优缺点自然也就各有不同了。

三种消息传输方式QoS：

0代表“至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。

1代表“至少一次”，确保消息到达，但消息重复可能会发生。

2代表“只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。 备注：由于服务端采用Mosca实现，Mosca目前只支持到QoS 1

如果发送的是临时的消息，例如给某topic所有在线的设备发送一条消息，丢失的话也无所谓，0就可以了（客户端登录的时候要指明支持的QoS级别，同时发送消息的时候也要指明这条消息支持的QoS级别），如果需要客户端保证能接收消息，需要指定QoS为1，如果同时需要加入客户端不在线也要能接收到消息，那么客户端登录的时候要指定session的有效性，接收离线消息需要指定服务端要保留客户端的session状态。

mqtt基于订阅者模型架构，客户端如果互相通信，必须在同一订阅主题下，即都订阅了同一个topic，客户端之间是没办法直接通讯的。订阅模型显而易见的好处是群发消息的话只需要发布到topic，所有订阅了这个topic的客户端就可以接收到消息了。

发送消息必须发送到某个topic，重点说明的是不管客户端是否订阅了该topic都可以向topic发送了消息，还有如果客户端订阅了该主题，那么自己发送的消息也会接收到。

小型传输，开销很小（固定长度的头部是2字节），协议交换最小化，以降低网络流量。这就是为什么在介绍里说它非常适合“在物联网领域，传感器与服务器的通信，信息的收集”，要知道嵌入式设备的运算能力和带宽都相对薄弱，使用这种协议来传递消息再适合不过了。

使用Last Will和Testament特性通知有关各方客户端异常中断的机制。Last Will：即遗言机制，用于通知同一主题下的其他设备发送遗言的设备已经断开了连接。Testament：遗嘱机制，功能类似于Last Will 。

在云服务器上搭建了mqtt，为什么手机连接不上mqtt，要怎么做才能连接上？求求大神帮忙

MQTT协议是广泛应用的物联网协议，使用测试MQTT协议需要MQTT的代理。有两种方法使用MQTT服务，一是租用现成的MQTT服务器，如阿里云，百度云，华为云等公用的云平台提供的MQTT服务，使用公用的MQTT服务器的好处是省事，但如果仅仅用于测试学习还需要注册帐号，灵活性差些，有的平台还需要付费。另一方法是自己使用开源的MQTT组件来搭建。

MQTT服务器非常多，如apache的ActiveMQ，emtqqd，HiveMQ，Emitter，Mosquitto，Moquette等等。

这里介绍的是用轻量级的mosquitto开源项目来搭建一个属于自己的MQTT服务器。

第一步：需要安装一台linux主机，这不多介绍，可以使用真机安装也可以使用虚拟机安装。如果仅仅是自己测试使用都可以。

第二步：下载mosquitto需要的依赖

sudo apt-get install libssl-devsudo apt-get install uuid-devsudo apt-get install cmake

第三步：下载mosquitto并解压，现在mosquitto官网最新的版本是1.5.1

tar xzvf mosquitto-1.5.1.tar.gz

第四步：编译

cd mosquitto-1.5.1/

make

make install

第五步：启动mosquitto

./mosquitto -v

1535473957: mosquitto version 1.5.1 starting

1535473957: Using default config.

1535473957: Opening ipv4 listen socket on port 1883.

1535473957: Opening ipv6 listen socket on port 1883.

这时候mosquitto就会以默认的参数启动。如果需要带配置文件可以修改配置文件mosquitto.conf，

启动时候加上参数 -c，

./mosquitto -c mosquitto.conf

可以看到，mosquitto监听的端口为1883.

这时候我们的MQTT服务器就搭建好了。可找一个mqtt客户端来测试一下。

先发布一个主题“home/garden/fountain/2”

内容是“hello world”

这时候在mosquitto会打印出下面的log

535474247: New connection from 192.168.1.105 on port 1883.

1535474247: New client connected from 192.168.1.105 as MQTT_FX_Client (c1, k60).

1535474247: No will message specified.

1535474247: Sending CONNACK to MQTT_FX_Client (0, 0)

1535474307: Received PINGREQ from MQTT_FX_Client

1535474307: Sending PINGRESP to MQTT_FX_Client

1535474339: Received PUBLISH from MQTT_FX_Client (d0, q0, r0, m0, 'home/garden/fountain/2', ... (12 bytes))

1535474367: Received PINGREQ from MQTT_FX_Client

1535474367: Sending PINGRESP to MQTT_FX_Client

订阅主题“home/garden/fountain/2”

可以看到收到了自己发布的消息。

用wireshark抓包

可以看到抓到了一个MQTT的publish的报文。

阿里云消息中间件（MQ）探秘

阅读字数： 2513 | 5分钟阅读

获取嘉宾演讲视频及PPT ，请点击：

阿里巴巴中间技术专家不铭从功能特性、技术架构、最佳实践、案例分析四个方面进行了《Aliware-MQ消息队列》的分享。

Aliware-MQ是阿里云提供的企业级互联网架构的核心产品，基于高可用分布式集群技术，支持海量高并发和万亿级消息流转，支持海量的消息堆积，支持高可靠/高可用方案，提供了运维、监控等一系列完整的配套服务。

如上图所示，从消息的维度来看分为普通消息、顺序消息、定时消息和事务消息等四种消息,无论是发送哪种消息客户端都支持熔断机制，即如果发现发送目标节点有性能问题，客户端会自动进行熔断，把有问题的节点排出去，保证消息发往可靠性最高的机器。管理方面已经支持消息的查询、消息回溯、消息全链路轨迹和监控报警机制。性能上MQ已经达到了百亿级的堆积能力，毫秒级的投递延迟，支持万级节点高并发，集群水平热扩缩。消息消费方面，支持失败后的消息重投机制，失败的消息会重新投递到队列中去，现在最多支持16次重投。

上图是Aliware-MQ的功能架构。左边是控制台的管理，可以在上面做发布订阅管理。右边目前的接入方式是SDK支持TCP协议，同时也支持HTTP接口，以及面向手机终端的MQTT协议。

OpenAPI是MQ提供给用户的管控方式，用于实现一系列资源管理和运维功能，用户可以通过Open API查询所需要的任何东西。

上图中是我们今年推出的一个MQ移动物联网套件。之前的客户端，不管是上游还是下游收发都是用各自的服务器。但是今年我们有了移动物联网套件，可以直接面向终端设备。比如手机、汽车等移动设备利用移动物联网套件，通过一个网关就可以直接和消息系统打通。

Aliware-MQ的消息系统是基于队列。队列要保证数据安全，是支持高并发和高性能读写的最基本元素。

如上图所示，Producer是消息发送集群，下游的Consumer是消费者集群，都依赖于MQ的SDK。Broker是消息服务器，所有的消息都发送到Broker上面；Name Server和ZK功能类似，用来做服务发现。Producer要从Name Server获取到Topic在哪个节点上，订阅Topic时需要知道Topic从哪里取，同样需要Name Server。Broker上的Topic信息会定时在Name Server上注册，Producer和Consumer在交互之前会从Name Server上获取目标。

图中的master是主机，slave是备机，主备之间会做数据同步，有异步和同步两种方式。一个master可以布多个节点，这个根据自己的成本来决定。如果扩容的话，只要直接布一台master即可，它会定时地将Topic注册到Name Server上，发送方和订阅方也会定时地感知这个过程，整个扩容的过程对于用户来说大概30秒就能完成。

Aliware-MQ所有数据存储在Commit Log里，它在实现上就相当于一个文件夹，每次会生成一个1G的文件。不管哪个Topic写过来的消息都会直接写入这个文件中，这个文件写满后再直接写下一个。

针对每一个Topic，要在业务层面对它进行区分，所以我们做了一层索引。例如在上图中有5个队列，每个队列都会生成定长的索引文件，通过索引，可以找到这条消息当前处于哪个CommitLog文件的某个具体位置中。

这样存储结构，保证了无论多少个topic，CommitLog的写是顺序的，能较大的保证MQ的写入性能。

Aliware-MQ的负载均衡是按照队列维度来做的，消费的时候会把topic的队列平均分配给消费实例。比如有2个消费实例，topic队列是4个，那么每个消费实例就消费2个；而如果共有5个队列，那么就是是1个消费2个，另1个消费3个。一个队列同一时间只会被一个消费实例消费，所以当出现队列数量小于消费实例数量的情况时，就会有消费实例出现空闲，这个时候可以根据业务实际情况手动通过工具将队列数量调大。

消息写进来都是先放在Java堆里，然后再落盘。如果用户要消费的消息都在内存里，那么就可以很快的读取到。但是如果用户消息堆积比较久，消息已经不在内存里而是存储在了磁盘中，这个时候就需要去磁盘里取数据，然后加载到内存里面读取出来。

Aliware-MQ的刷盘策略有异步和同步两种。异步到内存就返回成功，同步写则一定是消息刷到磁盘中才会返回成功。这种刷盘方式可以根据业务的具体需求进行配置，从写入的性能来看，异步写的性能肯定是会比同步的好。

从发消息的角度来看，如果发送失败，会有补偿机制。MQ的客户端会做三次重发，一台机器发送失败之后会默认往另外两台机器再尝试，如果三次都失败了才会把最终的失败结果传回，这个时候用户需要自己对发送异常进行相关处理。

有幂等要求的业务，Consumer在使用的时候需要自己做去重操作，在一些场景下，如客户端本地等待超时等，是无法保证消息完全不重复的，因此用户在进行系统设计时需要考虑到这一点。

Aliware-MQ目前支持的消息最大是4M，消息越小，性能越高。定时消息是支持消息的定时投递，可以自行设置要投递的时间，最长是40天。事务消息通过两阶段的提交的方式，来解决分布式事务问题。顺序消息可以采用全局顺序、分区顺序，严格保证消息的顺序。

Aliware-MQ的使用场景主要有系统间异步解耦、分布式事务、异构数据复制与分发、双十一大促的削峰填谷、大规模机器的Cache同步、日志服务和IM实时通信以及实时计算分析。

MQ顺序消息分为全局有序和队列有序。全局有序是从指所有消息发出开始，下游的接收方都是按照顺序接收；队列有序则是将消息进行区块分区，同一个分区内的消息按照先入先出的顺序进行顺序消费，保证一个队列只会被一个进程消费。

当一个交易系统下单之后，会发一条消息到MQ，购物车接收消息把购物车里的状态清空。如果这时交易消息发送失败，购物车就无法清空，对于数据来说这就是一个脏数据。面对这种情况我们有事务消息可以解决这个问题，在交易开始时先发送一条半事务消息，然后交易系统开始下单，所有事情做完之后再提交半事务，这时只有主动提交成功，消息队列才会将这条消息实际发送给用户。如果交易下单过程失败，则可以主动回滚这条消息，购物车和交易系统之间可以做到没有脏数据。

双十一大促时，各个分会场会有玲琅满目的商品，每件商品的价格都会实时变化。使用缓存技术也无法满足对商品价格的访问需求，缓存服务器网卡跑满。访问较多次商品价格查询影响会场页面的打开速度。于是MQ提供了一种广播机制，本来一条消息只会被集群的一台机器消费。如果使用广播模式，那么这条消息会被集群下的所有节点消费一次，相当于把价格信息同步到需要的每台机器上，可以取代缓存的作用。

实时计算功能主要是做一个消息总线，业务系统自动采集数据，把消息分发达下游的实时计算系统里，根据实时计算结果来给业务方做服务。

我今天的分享就到这里，谢谢大家！

NE35E MQTT协议对接阿里云

MQTT通信协议的基本介绍参考文章 NT35 MQTT通信 ，本篇给出阿里云的基本操作，NT35E通过订阅阿里云的主题发布信息与阿里云平台相互通信。

登录阿里云 → 工作台 → 物联网平台 → 进入控制台→ 公共实例

阿里云默认通信协议为MQTT，不需要特殊选择，用户按照如下步骤创建自己的产品：

创建产品 → 添加设备 

在"查看"标签中，包含了MQTT连接的基本三元组信息，也就是后面设备要填充的基本参数

      用户每定义一类产品都会自动生成对应的Topic列表，当然我们也可以"自定义Topic"便于自己测试。

       指令解析参考《Lierda NT35ENT26E-CN AT命令手册》，这里给出使用到的AT指令对应参数说明以便于理解。

AT+LMQTTCFG=cloud,tcpconnectID[,cloud _ type,data_type ]

tcpconnectID 。MQTT Socket 标识符。范围：0~4。

cloud_type整型。2 alibaba，其他参数指定其他平台

data_type整型。阿里云平台 1 json数据 

AT+LMQTTCFG="cloud",0,2,1   对应就是对接阿里云平台，发送json格式的数据

AT+LMQTTCFG=aliauth,tcpconnectID[,product_key,device_name,device_secret]

填充阿里云平台中设备的三元组信息

AT+LMQTTCFG="aliauth",0,"a1JszCpjS61","NT35E_06011","390358fc595040aa73221e8393aba86c"

这部分是模组进行TCP链路连接(需抓包确认)

AT+LMQTTOPEN=tcpconnectID,host_name,port

host_name对应阿里云 "设备信息"→"MQTT连接参数" 中的 "mqttHostUrl"

AT+LMQTTOPEN=0,"a1JszCpjS61.iot-as-mqtt.cn-shanghai.aliyuncs.com",1883

模组作为客户端，通过MQTT协议连接到服务器(需抓包确认)

AT+LMQTTCONN=tcpconnectID[,clientID[,username[,password]]]

clientID字符串型。客户端标识符。用户可以随便定义。 username,password 不需要填写

AT+LMQTTCONN=0,"NT35E"

AT+LMQTTSUBUNSUB=tcpconnectID,subflag,msgID,topic1[,qos1[,topic2[,qos2]d…]]

subflag整型。消息类型 0 订阅 1 取消订阅

msgID整型。数据包消息标识符。范围：0~65535。

topic带双引号的字符串型。客户端订阅或者退订的主题。长度范围：0~256 字节。

qos整型。客户端发送订阅消息(SUBSCRIBE)的 QoS 等级，此时为必选参数。2 正好一次，该主题下的消息确保接收端仅接收到一次

AT+LMQTTSUBUNSUB=0,0,1,"/a1JszCpjS61/ NT35E_06011 /user/COMMUTEST",2

这里注意topic对应参数的替换，里面的deviceName需要替换。

订阅主题之后，服务器下发的数据模组就可以正常接收了。模组下发位置

       发布消息在对应的设备目录下，如果有设备"订阅"对应的消息，平台"发布"相应的数据设备就可以接收到了。

AT+LMQTTPUB=tcpconnectID,msgID,qos,retain,topic,msglen,msg

msgID整型。 0~65535。任意定义，但qos=0 时，该参数值只能为0。

qos整型。 0 最多一次 1 至少一次 2  正好一次

retain整型 。服务器是否保存该消息。0 不保存  1 保存

topic带双引号的字符串型。 客户端发布消息的主题。长度范围：0~256 字节

msglen整型 。指定的消息数据长度。范围：0~1460。

msg字符串型。 需要发布的消息数据。

AT+LMQTTPUB=0,0,0,1,"/a1JszCpjS61/ NT35E_06011 /user/COMMUTEST",10,"1122334455"

       注意刚刚自己创建的主题属性是" 发布和订阅 "，所以模组发送该主题的信息，阿里云也是可以收到的

注意这里模组发送数据的时候，也推送了自己发送的数据，因为刚刚订阅了这个主题，所以模组订阅(收)到了对应的数据

       前面我们通过NT35E与平台进行信息交互，那么为什么是这样填写对应的参数呢，每个参数对应的说明在阿里云上是什么样的呢，用户可以查看阿里云的帮助文档进行确认。

       上面我们使用三元组的方式( 一机一密 )实现NT35E与阿里云平台通信，但实际生产过程中该方式不好实现，比如工厂有1000个设备生产，如果每个设备都复制不同的三元组，很难实现工厂批量化生产，此时可以通过 一型一密 的通信方式解决该问题。

一型一密模组端实现方式后续更新。