MQTT以及mosquitto的学习_mqtt协议服务器与mosquitto

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

---

一、MQTT的相关介绍

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用

1.MQTT的特点

MQTT协议工作在低带宽、不可靠的网络的远程传感器和控制设备通讯而设计的协议，它具有以下主要的几项特性：

（1）使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。

这一点很类似于XMPP，但是MQTT的信息冗余远小于XMPP，,因为XMPP使用XML格式文本来传递数据。

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

（3）使用TCP/IP提供网络连接。

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

（4）有三种消息发布服务质量：

“至多一次”，消息发布完全依赖底层TCP/IP网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。这一种方式主要普通APP的推送，倘若你的智能设备在消息推送时未联网，推送过去没收到，再次联网也就收不到了。

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

“只有一次”，确保消息到达一次。在一些要求比较严格的计费系统中，可以使用此级别。在计费系统中，消息重复或丢失会导致不正确的结果。这种最高质量的消息发布服务还可以用于即时通讯类的APP的推送，确保用户收到且只会收到一次。

（5）小型传输，开销很小（固定长度的头部是2字节），协议交换最小化，以降低网络流量。

这就是为什么在介绍里说它非常适合"在物联网领域，传感器与服务器的通信，信息的收集"，要知道嵌入式设备的运算能力和带宽都相对薄弱，使用这种协议来传递消息再适合不过了。

（6）使用Last Will和Testament特性通知有关各方客户端异常中断的机制。

Last Will：即遗言机制，用于通知同一主题下的其他设备发送遗言的设备已经断开了连接。

Testament：遗嘱机制，功能类似于Last Will。

2.MQTT的协议原理

MQTT协议实现方式
实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

（1）Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）

（2）payload，可以理解为消息的内容，是指订阅者具体要使用的内容。

网络传输与应用消息

MQTT会构建底层网络传输：它将建立客户端到服务器的连接，提供两者之间的一个有序的、无损的、基于字节流的双向传输。

当应用数据通过MQTT网络发送时，MQTT会把与之相关的服务质量（QoS）和主题名（Topic）相关连。

MQTT客户端

一个使用MQTT协议的应用程序或者设备，它总是建立到服务器的网络连接。客户端可以：

（1）发布其他客户端可能会订阅的信息；
（2）订阅其它客户端发布的消息；
（3）退订或删除应用程序的消息；
（4）断开与服务器连接。

MQTT服务器

MQTT服务器以称为"消息代理"（Broker），可以是一个应用程序或一台设备。它是位于消息发布者和订阅者之间，它可以：

（1）接受来自客户的网络连接；
（2）接受客户发布的应用信息；
（3）处理来自客户端的订阅和退订请求；
（4）向订阅的客户转发应用程序消息。

MQTT协议中的订阅、主题、会话

一、订阅（Subscription）

订阅包含主题筛选器（Topic Filter）和最大服务质量（QoS）。订阅会与一个会话（Session）关联。一个会话可以包含多个订阅。每一个会话中的每个订阅都有一个不同的主题筛选器。

二、会话（Session）

每个客户端与服务器建立连接后就是一个会话，客户端和服务器之间有状态交互。会话存在于一个网络之间，也可能在客户端和服务器之间跨越多个连续的网络连接。

三、主题名（Topic Name）

连接到一个应用程序消息的标签，该标签与服务器的订阅相匹配。服务器会将消息发送给订阅所匹配标签的每个客户端。

四、主题筛选器（Topic Filter）

一个对主题名通配符筛选器，在订阅表达式中使用，表示订阅所匹配到的多个主题。

五、负载（Payload）

消息订阅者所具体接收的内容。

MQTT协议中的方法

MQTT协议中定义了一些方法（也被称为动作），来于表示对确定资源所进行操作。这个资源可以代表预先存在的数据或动态生成数据，这取决于服务器的实现。通常来说，资源指服务器上的文件或输出。主要方法有：

（1）Connect。等待与服务器建立连接。
（2）Disconnect。等待MQTT客户端完成所做的工作，并与服务器断开TCP/IP会话。
（3）Subscribe。等待完成订阅。
（4）UnSubscribe。等待服务器取消客户端的一个或多个topics订阅。
（5）Publish。MQTT客户端发送消息请求，发送完成后返回应用程序线程。

3.MQTT协议数据包结构

在MQTT协议中，一个MQTT数据包由：固定头（Fixed header）、可变头（Variable header）、消息体（payload）三部分构成。MQTT数据包结构如下：

（1）固定头（Fixed header）。存在于所有MQTT数据包中，表示数据包类型及数据包的分组类标识。
（2）可变头（Variable header）。存在于部分MQTT数据包中，数据包类型决定了可变头是否存在及其具体内容。
（3）消息体（Payload）。存在于部分MQTT数据包中，表示客户端收到的具体内容。

MQTT固定头

固定报头，所有的MQTT控制报文都包含，可变报头与有效载荷是部分MQTT控制报文包含。 固定报头占据两字节的空间，具体见

固定报头的第一个字节分为控制报文的类型（4bit），以及控制报文类型的标志位，控制类型共有14种，其中0与15被系统保留出来，其他的类型具体见：

固定报头的bit0-bit3为标志位，依照报文类型有不同的含义，事实上，除了PUBLISH类型报文以外，其他报文的标志位均为系统保留，PUBLISH报文的第一字节bit3是控制报文的重复分发标志（DUP），bit1-bit2是服务质量等级，bit0是PUBLISH报文的保留标志，用于标识PUBLISH是否保留，当客户端发送一个PUBLISH消息到服务器，如果保留标识位置1，那么服务器应该保留这条消息，当一个新的订阅者订阅这个主题的时候，最后保留的主题消息应被发送到新订阅的用户。

固定报头的第二个字节开始是剩余长度字段，是用于记录剩余报文长度的，表示当前的消息剩余的字节数，包括可变报头和有效载荷区域（如果存在），但剩余长度不包括用于编码剩余长度字段本身的字节数。

剩余长度字段使用一个变长度编码方案，对小于128的值它使用单字节编码，而对于更大的数值则按下面的方式处理：每个字节的低7位用于编码数据长度，最高位（bit7）用于标识剩余长度字段是否有更多的字节，且按照大端模式进行编码，因此每个字节可以编码128个数值和一个延续位，剩余长度字段最大可拥有4个字节。

当剩余长度使用1个字节存储时，其取值范围为0(0x00)~127(0x7f)。 
 
当使用2个字节时，其取值范围为128(0x80,0x01)~16383(0Xff,0x7f)。
 
当使用3个字节时，其取值范围为16384(0x80,0x80,0x01)~2097151(0xFF,0xFF,0x7F)。 
 
当使用4个字节时，其取值范围为2097152(0x80,0x80,0x80,0x01)~268435455(0xFF,0xFF,0xFF,0x7F)

总的来说，MQTT报文理论上可以发送最大256M的报文，当然，这种情况是非常少的。
固定头存在于所有MQTT数据包中，其结构如下：

MQTT数据包类型

位置：Byte 1中bits 7-4。

相于一个4位的无符号值，类型、取值及描述如下：

标识位

位置：Byte 1中bits 3-0。

在不使用标识位的消息类型中，标识位被作为保留位。如果收到无效的标志时，接收端必须关闭网络连接：

（1）DUP：发布消息的副本。用来在保证消息的可靠传输，如果设置为1，则在下面的变长中增加MessageId，并且需要回复确认，以保证消息传输完成，但不能用于检测消息重复发送。

（2）QoS：发布消息的服务质量，即：保证消息传递的次数

Ø00：最多一次，即：<=1
 
Ø01：至少一次，即：>=1
        
Ø10：一次，即：=1
 
Ø11：预留

（3）RETAIN： 发布保留标识，表示服务器要保留这次推送的信息，如果有新的订阅者出现，就把这消息推送给它，如果设有那么推送至当前订阅者后释放。

剩余长度（Remaining Length）

地址：Byte 2。

固定头的第二字节用来保存变长头部和消息体的总大小的，但不是直接保存的。这一字节是可以扩展，其保存机制，前7位用于保存长度，后一部用做标识。当最后一位为1时，表示长度不足，需要使用二个字节继续保存。例如：计算出后面的大小为0

MQTT可变头

MQTT数据包中包含一个可变头，它驻位于固定的头和负载之间。可变头的内容因数据包类型而不同，较常的应用是作为包的标识：

只有某些报文才拥有可变报头，它在固定报头和有效负载之间，可变报头的内容会根据报文类型的不同而有所不同，但可变报头的报文标识符（Packet Identifier）字段存在于在多个类型的报文里，而有一些报文又没有报文标识符字段，具体见表格，报文标识符结构具体见图：

Payload消息体

Payload消息体位MQTT数据包的第三部分，包含CONNECT、SUBSCRIBE、SUBACK、UNSUBSCRIBE四种类型的消息：

（1）CONNECT，消息体内容主要是：客户端的ClientID、订阅的Topic、Message以及用户名和密码。
（2）SUBSCRIBE，消息体内容是一系列的要订阅的主题以及QoS。
（3）SUBACK，消息体内容是服务器对于SUBSCRIBE所申请的主题及QoS进行确认和回复。
（4）UNSUBSCRIBE，消息体内容是要订阅的主题。

 二：mosquitto的相关介绍

1.mosquitto是什么

mosquitto是一款实现了消息推送协议 MQTT v3.1 的开源消息代理软件，提供轻量级的，支持可发布/可订阅的的消息推送模式，使设备对设备之间的短消息通信变得简单。
在实验中使用mosquitto库函数来实现订阅与发布。

2.安装mosquitto

1.sudo命令安装

# 安装mosquitto
sudo apt-get install mosquitto
# 安装客户端
sudo apt-get install mosquitto-clients
# 安装设备端
sudo apt-get install mosquitto-dev
#查询mosquitto是否正确运行
mosquitto -p 1883   //mosquitto默认时候用1883端口

2.源码安装

（1）wget下载源码包

wget http://mosquitto.org/files/source/mosquitto-1.5.5.tar.gz

（2)解压

tar -xzvf mosquitto-1.5.5.tar.gz

(3)进入目录

cd mosquitto-1.5.5/

(4)编译

make

(5)运行安装

sudo make install

可能出现问题及解决方法

【1】编译找不到openssl/ssl.hsudo apt-get install libssl-dev

【2】编译过程g++命令未找到：sudo apt-get install g++

【3】编译过程找不到ares.hsudo apt-get install libc-ares-dev

【4】编译过程找不到uuid/uuid.hsudo apt-get install uuid-dev

【5】使用过程中找不到libmosquitto.so.1 error while loading shared libraries: libmosquitto.so.1: cannot open shared object file: No such file or directory
【解决方法】——修改libmosquitto.so位置
创建链接sudo ln -s /usr/local/lib/libmosquitto.so.1 /usr/lib/libmosquitto.so.1
更新动态链接库sudo ldconfig
参考：MQTT简介和mosquitto安装与使用及利用mosquitto库Linux c编程_WuYuJun's blog的博客-CSDN博客

3.mosquitto命令

mosquitto [-c config file] [ -d | --daemon ] [-p port number] [-v]

参数说明：
-c 后面跟的是启动mosquitto可以调整的参数，比如是否开启基本认证，端口是什么，SSL单向和双向的认证配置等等。
-d 表示MQTT mosquitto将在后台运行。
-p 代表当前的mosquitto服务实例启动以后，其监听端口号，这个配置覆盖[-c config file] 指定的配置文件中的端口。
-v 代码调试模式（verbose）在终端输出更多的信息 。

订阅端：

mosquitto_sub 命令参数说明
1. -c  设定‘clean session’为无效状态，这样一直保持订阅状态，即便是已经失去连接，如果再次连接仍旧能够接收的断开期间发送的消息。
2. -d  打印debug信息
3. -h  指定要连接的域名  默认为localhost
4. -i 指定clientId
5. -I 指定clientId前缀
6. -k keepalive 每隔一段时间，发PING消息通知broker，仍处于连接状态。 默认为60秒。
7. -q 指定希望接收到QoS为什么的消息  默认QoS为0
8. -R 不显示陈旧的消息
9. -t 订阅topic
10. -v 打印消息
11. --will-payload  指定一个消息，该消息当客户端与broker意外断开连接时发出。该参数需要与--will-topic一起使用
12. --will-qos  Will的QoS值。该参数需要与--will-topic一起使用
13. --will-retain 指定Will消息被当做一个retain消息（即消息被广播后，该消息被保留起来）。该参数需要与--will-topic一起使用
14. --will-topic  用户发送Will消息的topic

发布端：

mosquitto_pub 命令参数说明
1. -d  打印debug信息
2. -f  将指定文件的内容作为发送消息的内容
3. -h  指定要连接的域名  默认为localhost
4. -i  指定要给哪个clientId的用户发送消息
5. -I  指定给哪个clientId前缀的用户发送消息
6. -m  消息内容
7. -n  发送一个空（null）消息
8. -p  连接端口号
9. -q  指定QoS的值（0,1,2）
10. -t  指定topic
11. -u  指定broker访问用户
12. -P  指定broker访问密码
13. -V  指定MQTT协议版本
14. --will-payload  指定一个消息，该消息当客户端与broker意外断开连接时发出。该参数需要与--will-topic一起使用
15. --will-qos  Will的QoS值。该参数需要与--will-topic一起使用
16. --will-retain 指定Will消息被当做一个retain消息（即消息被广播后，该消息被保留起来）。该参数需要与--will-topic一起使用
17. --will-topic  用户发送Will消息的topic

4.mosquitto库函数

int mosquitto_lib_init(void)
//功能：使用mosquitto库函数前，要先初始化，使用之后就要清除。清除函数；int mosquitto_lib_cleanup();
//返回值：MOSQ_ERR_SUCCESS 总是
 

int mosquitto_lib_cleanup（void）
//功能：使用完mosquitto函数之后，要做清除工作。
//返回值： MOSQ_ERR_SUCCESS 总是

struct mosquitto *mosquitto_new( const char * id, bool clean_session, void * obj )
 
/*功能：创建一个新的mosquitto客户端实例，新建客户端。
参数：①id ：用作客户端ID的字符串。如果为NULL，将生成一个随机客户端ID。如果id为NULL，clean_session必须为true。
②clean_session：设置为true以指示代理在断开连接时清除所有消息和订阅，设置为false以指示其保留它们，客户端将永远不会在断开连接时丢弃自己的传出消息。调用mosquitto_connect或mosquitto_reconnect将导致重新发送消息。使mosquitto_reinitialise将客户端重置为其原始状态。如果id参数为NULL，则必须将其设置为true。
简言之：就是断开后是否保留订阅信息true/false
③obj： 用户指针，将作为参数传递给指定的任何回调，（回调参数）
返回：成功时返回结构mosquitto的指针，失败时返回NULL，询问errno以确定失败的原因：
ENOMEM内存不足。
EINVAL输入参数无效。

void mosquitto_destroy( struct mosquitto * mosq )
 
/*功能：释放客户端
参数：mosq： struct mosquitto指针

void mosquitto_connect_callback_set(struct mosquitto * mosq, void (*on_connect)(struct mosquitto *mosq, void *obj, int rc) )
 
/*功能：设置连接确认时回调函数，当代理发送CONNACK消息以响应连接时，将调用on_connect。
参数：mosq： struct mosquitto指针
void (*on_connect)(struct mosquitto *mosq , void *obj, int rc)  回调函数（参数：
mosq： struct mosquitto指针
obj：mosquitto_new中提供的用户数据
rc： 连接响应的返回码，其中有：
0-成功
1-连接被拒绝（协议版本不可接受）
2-连接被拒绝（标识符被拒绝）
3-连接被拒绝（经纪人不可用）
4-255-保留供将来使用
）

void mosquitto_disconnect_callback_set( struct mosquitto *mosq,void (*on_disconnect)( struct mosquitto *mosq,void *obj, int rc) );
 
/*功能：设置断开连接回调函数，当代理收到DISCONNECT命令并断开与客户端的连接，将调用on_disconnect。
参数：mosq： struct mosquitto指针
void (*on_connect)(struct mosquitto *mosq , void *obj, int rc)  回调函数 （参数：
mosq： struct mosquitto指针
obj：mosquitto_new中提供的用户数据
rc：0表示客户端已经调用mosquitto_disconnect，任何其他值，表示断开连接时意外的。）

int mosquitto_connect( struct mosquitto * mosq, const char * host, int port, int keepalive )
 
/*功能: 连接到MQTT代理/服务器（主题订阅要在连接服务器之后进行）
参数：①mosq ： 有效的mosquitto实例，mosquitto_new（）返回的mosq.
②host : 服务器ip地址
③port：服务器的端口号
④keepalive：保持连接的时间间隔， 单位秒。如果在这段时间内没有其他消息交换，则代理应该将PING消息发送到客户端的秒数。
返回：MOSQ_ERR_SUCCESS 成功。
MOSQ_ERR_INVAL 如果输入参数无效。
MOSQ_ERR_ERRNO 如果系统调用返回错误。变量errno包含错误代码

int mosquitto_disconnect( struct mosquitto * mosq )
 
/*功能：断开与代理/服务器的连接。
返回：
MOSQ_ERR_SUCCESS 成功。
MOSQ_ERR_INVAL 如果输入参数无效。
MOSQ_ERR_NO_CONN 如果客户端未连接到代理。

int mosquitto_publish( struct mosquitto * mosq, int * mid, const char * topic, int payloadlen, const void * payload, int qos, bool retain )
 
/*功能：主题发布的函数
参数：①mosq：有效的mosquitto实例，客户端
②mid：指向int的指针。如果不为NULL，则函数会将其设置为该特定消息的消息ID。然后可以将其与发布回调一起使用，以确定何时发送消息。请注意，尽管MQTT协议不对QoS = 0的消息使用消息ID，但libmosquitto为其分配了消息ID，以便可以使用此参数对其进行跟踪。
③topic：要发布的主题，以null结尾的字符串
④payloadlen：有效负载的大小（字节），有效值在0到268，435，455之间；主题消息的内容长度
⑤payload： 主题消息的内容，指向要发送的数据的指针，如果payloadlen >0，则它必须时有效的存储位置。
⑥qos：整数值0、1、2指示要用于消息的服务质量。
⑦retain：设置为true以保留消息。
返回:
MOSQ_ERR_SUCCESS 成功。
MOSQ_ERR_INVAL 如果输入参数无效。
MOSQ_ERR_NOMEM 如果发生内存不足的情况。
MOSQ_ERR_NO_CONN 如果客户端未连接到代理。
MOSQ_ERR_PROTOCOL 与代理进行通信时是否存在协议错误。
MOSQ_ERR_PAYLOAD_SIZE 如果payloadlen太大。
MOSQ_ERR_MALFORMED_UTF8 如果主题无效，则为UTF-8
MOSQ_ERR_QOS_NOT_SUPPORTED 如果QoS大于代理支持的QoS。
MOSQ_ERR_OVERSIZE_PACKET 如果结果包大于代理支持的包。

int mosquitto_subscribe( struct mosquitto * mosq, int * mid, const char * sub, int qos )
/*功能：订阅主题函数
参数：①mosq：有效的mosquitto实例，客户端
②mid： 指向int的指针。如果不为NULL，则函数会将其设置为该特定消息的消息ID。然后可以将其与订阅回调一起使用，以确定何时发送消息。；主题的消息ID
③sub： 主题名称，订阅模式。
④qos : 此订阅请求的服务质量。
返回值：
MOSQ_ERR_SUCCESS 成功。
MOSQ_ERR_INVAL 如果输入参数无效。
MOSQ_ERR_NOMEM 如果发生内存不足的情况。
MOSQ_ERR_NO_CONN 如果客户端未连接到代理。
MOSQ_ERR_MALFORMED_UTF8 如果主题无效，则为UTF-8
MOSQ_ERR_OVERSIZE_PACKET 如果结果包大于代理支持的包。

void mosquitto_message_callback_set( struct mosquitto * mosq, void (*on_message)(struct mosquitto *, void *, const struct mosquitto_message *) )
 
/*功能：设置消息回调函数，收到订阅的消息后调用on_message。
参数：①mosq： 有效的mosquitto实例，客户端。
②on_message 回调函数，格式如下：void callback（struct mosquitto * mosq，void * obj，const struct mosquitto_message * message）
回调的参数：
①mosq：进行回调的mosquitto实例
②obj： mosquitto_new中提供的用户数据
③message: 消息数据，回调完成后，库将释放此变量和关联的内存，客户应复制其所需要的任何数据。
struct mosquitto_message{
int mid;//消息序号ID
char *topic; //主题
void *payload; //主题内容 ，MQTT 中有效载荷
int payloadlen; //消息的长度，单位是字节
int qos; //服务质量
bool retain; //是否保留消息
};

int mosquitto_loop_forever( struct mosquitto * mosq, int timeout, int max_packets )
 
/*功能：此函数在无限阻塞循环中为你调用loop（），对于只想在程序中运行MQTT客户端循环的情况，这很有用，如果服务器连接丢失，它将处理重新连接，如果在回调中调用mosqitto_disconnect（）它将返回。
参数：①mosq: 有效的mosquitto实例，客户端
②timeout： 超时之前，在select（）调用中等待网络活动的最大毫秒数，设置为0以立即返回，设置为负可使用默认值为1000ms。
③max_packets： 该参数当前未使用，应设为为1，以备来兼容
返回值：
MOSQ_ERR_SUCCESS 成功。
MOSQ_ERR_INVAL 如果输入参数无效。
MOSQ_ERR_NOMEM 如果发生内存不足的情况。
MOSQ_ERR_NO_CONN 如果客户端未连接到代理。
MOSQ_ERR_CONN_LOST 如果与代理的连接丢失。
MOSQ_ERR_PROTOCOL 与代理进行通信时是否存在协议错误。
MOSQ_ERR_ERRNO 如果系统调用返回错误。变量errno包含错误代码

int mosquitto_loop_stop( struct mosquitto * mosq, bool force )
 
/*功能:网络事件阻塞回收结束处理函数，这是线程客户端接口的一部分。调用一次可停止先前使用mosquitto_loop_start创建的网络线程。该调用将一直阻塞，直到网络线程结束。为了使网络线程结束，您必须事先调用mosquitto_disconnect或将force参数设置为true。
参数：①mosq :有效的mosquitto实例
②force：设置为true强制取消线程。如果为false，则必须已经调用mosquitto_disconnect。
返回：
MOSQ_ERR_SUCCESS 成功。
MOSQ_ERR_INVAL 如果输入参数无效。
MOSQ_ERR_NOT_SUPPORTED 如果没有线程支持。

int mosquitto_loop_start( struct mosquitto * mosq )
/*功能：网络事件循环处理函数，通过创建新的线程不断调用mosquitto_loop() 函数处理网络事件，不阻塞
返回：
MOSQ_ERR_SUCCESS 成功。
MOSQ_ERR_INVAL 如果输入参数无效。
MOSQ_ERR_NOT_SUPPORTED 如果没有线程支持。

附相关API链接：mosquitto.h