15 MQTT协议(二)

嵌入式Linux开发 APP
1

16.3 报文分析

16.3.1 CONNECT-连接服务端

​ 客户端到服务端的网络连接建立（完成三次握手）后，客户端发送给服务端的第一个报文必须是 CONNECT 报文。

image
image820×732 62.5 KB

​ 在一个网络连接上，客户端只能发送一次 CONNECT 报文。服务端必须将客户端发送的第二个 CONNECT报文当作协议违规处理并断开客户端的连接。

​ 有效载荷包含一个或多个编码的字段。 包括客户端的唯一标识符， Will 主题， Will 消息， 用户名和密码。 除了客户端标识之外， 其它的字段都是可选的， 基于标志位来决定可变报头中是否需要包含这些字段。

image
image878×935 124 KB

16.3.1.1 connect固定报头

bit 7 6 5 4 3 2 1 0
Byte1 Mqtt报文类型(1) Reserved(保留位)
0 0 0 1 0 0 0 0
Byte2~n 剩余长度

表格3.1

16.3.1.2 协议名字节组成

说明 7 6 5 4 3 2 1 0
协议名
Byte1 协议名长度MSB(0) 0 0 0 0 0 0 0 0
Byte2 协议名长度LSB(4) 0 0 0 0 0 1 0 0
Byte3 ‘M’ 0 1 0 0 1 1 0 1
Byte4 ‘Q’ 0 1 0 1 0 0 0 1
Byte5 ‘T’ 0 1 0 1 0 1 0 0
Byte6 ‘T’ 0 1 0 1 0 1 0 0

​ 数据包检测工具， 例如防火墙， 可以使用协议名来识别 MQTT 流量。

16.3.1.3 协议级别

说明 7 6 5 4 3 2 1 0
协议级别
Byte7 Level(4) 0 0 0 0 0 1 0 0

​ 客户端用 8 位的无符号值表示协议的修订版本。对于 3.1.1 版协议，协议级别字段的值是 4(0x04)。如果发现不支持的协议级别，服务端必须给发送一个返回码为 0x01（不支持的协议级别）的CONNACK 报文响应CONNECT 报文， 然后断开客户端的连接。

16.3.1.4 连接标记

bit 7 6 5 4 3 2 1 0
用户名标记 用户密码标记 Will retain Will qos Will flag 清除会话 reserved
Byte 8 x x x x x x 0

bit1清除会话

一般来说， 客户端连接时总是将清理会话标志设置为 0 或 1， 并且不交替使用两种值。 这个选择取决于具体的应用。 清理会话标志设置为 1 的客户端不会收到旧的应用消息， 而且在每次连接成功后都需要重新订阅任何相关的主题。清理会话标志设置为 0 的客户端会收到所有在它连接断开期间发布的 QoS 1 和 QoS 2 级别的消息。因此， 要确保不丢失连接断开期间的消息， 需要使用 QoS 1 或QoS 2 级别，同时将清理会话标志设置为 0。

Bit2遗嘱标志

遗嘱标志（Will Flag） 被设置为 1，表示如果连接请求被接受了， 遗嘱（Will Message） 消息必须被存储在服务端并且与这个网络连接关联。之后网络连接关闭时，服务端必须发布这个遗嘱消息， 除非服务端收到DISCONNECT 报文时删除了这个遗嘱消息。

Bit3和 bit4遗嘱 QoS

这两位用于指定发布遗嘱消息时使用的服务质量等级, 如果遗嘱标志被设置为 0， 遗嘱 QoS 也必须设置为 0(0x00)，如果遗嘱标志被设置为 1， 遗嘱 QoS 的值可以等于 0(0x00)， 1(0x01)， 2(0x02)， 它的值不能等于 3。

Bit5遗嘱保留

如果遗嘱消息被发布时需要保留，需要指定这一位的值, 如果遗嘱标志被设置为 0， 遗嘱保留（Will Retain） 标志也必须设置为 0 。
如果遗嘱标志被设置为 1：
· 如果遗嘱保留被设置为 0， 服务端必须将遗嘱消息当作非保留消息发布 。
· 如果遗嘱保留被设置为 1， 服务端必须将遗嘱消息当作保留消息发布。

Bit7 用户名标志

如果用户名（User Name） 标志被设置为 0， 有效载荷中不能包含用户名字段。

如果用户名（User Name） 标志被设置为 1， 有效载荷中必须包含用户名字段。

Bit6 用户名密码标记

如果密码（Password） 标志被设置为 0， 有效载荷中不能包含密码字段 。
如果密码（Password） 标志被设置为 1， 有效载荷中必须包含密码字段 。
如果用户名标志被设置为 0， 密码标志也必须设置为 0 。

16.3.1.5 保持连接

bit 7 6 5 4 3 2 1 0
Byte9 保持连接 Keep Alive MSB
Byte10 保持连接 Keep Alive LSB

a) 保持连接（Keep Alive） 是一个以秒为单位的时间间隔，表示为一个 16 位的字，它是指在客户端传输完成。

b) 一个控制报文的时刻到发送下一个报文的时刻， 两者之间允许空闲的最大时间间隔。 客户端负责保证控制。

c) 报文发送的时间间隔不超过保持连接的值。 如果没有任何其它的控制报文可以发送， 客户端必须发送一个PINGREQ 报文。

d) 不管保持连接的值是多少，客户端任何时候都可以发送 PINGREQ 报文，并且使用 PINGRESP 报文判断网络和服务端的活动状态。

e) 如果保持连接的值非零，并且服务端在一点五倍的保持连接时间内没有收到客户端的控制报文， 它必须断开客户端的网络连接， 认为网络连接已断开。

f) 客户端发送了 PINGREQ 报文之后， 如果在合理的时间内仍没有收到 PINGRESP 报文， 它应该关闭到服务端的网络连接。

g) 保持连接的值为零表示关闭保持连接功能。 这意味着，服务端不需要因为客户端不活跃而断开连接。 注意：不管保持连接的值是多少， 任何时候，只要服务端认为客户端是不活跃或无响应的， 可以断开客户端的连接。

16.3.1.6 客户端标识符

​ 服务端使用客户端标识符 (ClientId) 识别客户端。 连接服务端的每个客户端都有唯一的客户端标识符（ClientId） 。客户端和服务端都必须使用 ClientId 识别两者之间的 MQTT 会话相关的状态, 客户端标识符 (ClientId) 必须存在而且必须是 CONNECT 报文有效载荷的第一个字段,客户端标识符必须是UTF-8 编码字符串。

16.3.1.7 遗嘱主题

​ 如果遗嘱标志被设置为 1， 有效载荷的下一个字段是遗嘱主题（Will Topic） 。 遗嘱主题必须是 UTF-8 编码字符串。

16.3.1.8 遗嘱消息

​ 如果遗嘱标志被设置为 1， 有效载荷的下一个字段是遗嘱消息。 遗嘱消息定义了将被发布到遗嘱主题的应用消息。

16.3.1.9 用户名和密码

​ 如果用户名（ User Name） 标志被设置为 1， 有效载荷的下一个字段就是它。 用户名必须是定义的UTF-8 编码字符串。服务端可以将它用于身份验证和授权。

​ 如果密码（ Password） 标志被设置为 1， 有效载荷的下一个字段就是它。密码字段包含一个两字节的长度字段， 长度表示二进制数据的字节数（ 不包含长度字段本身占用的两个字节），后面跟着 0 到 65535 字节的二进制数据。
image

16.3.10.1 wirshark抓包分析connect报文

​ 从抓包可知，从上到下分别是固定报头，可变报头，连接标记，保持连接，用户名，用名密码，其中没有遗嘱相关消息字段，与3.1.1节分析的固定报头组成分析一致。

image
image1646×914 60.9 KB

16.3.10.2 c语言构造mqtt connect报文

static uint8_t client_id[512] = {"mqtt_client"};
static uint8_t user_name[512] = {"mqtt"};
static uint8_t passwd[512] = {"12345678"};
#define KEEP_ALIVE 20
int mqtt_connect(int sockfd)
{
    uint8 flags = 0x00;
	uint8 *packet = NULL;
	uint16 packet_length = 0;	
	uint16 clientidlen = strlen(client_id);
	uint16 usernamelen = strlen(user_name);
	uint16 passwordlen = strlen(passwd);
	uint16 payload_len = clientidlen + 2;
	// Variable header
	uint8 var_header[10] = {
		0x00,0x04,/*len*/
        0x4d,0x51,0x54,0x54,/*mqtt*/
		0x04,/*协议版本*/};
	
	uint8 fixedHeaderSize = 2;    // Default size = one byte Message Type + one byte Remaining Length
	uint8 remainLen = 0;
	uint8 *fixed_header = NULL;
	uint16 offset = 0;
    
	// Preparing the flags
	if(usernamelen) { /*用户名长度(可选)*/
		payload_len += usernamelen + 2;
		flags |= MQTT_USERNAME_FLAG;/*或上用户名标记*/
	}
	if(passwordlen) { /*用户密码(可选)*/
		payload_len += passwordlen + 2;
		flags |= MQTT_PASSWORD_FLAG;/*用户密码标记位*/
	}
	flags |= MQTT_CLEAN_SESSION;
	var_header[7] = flags;/*连接标记*/
	var_header[8] = KEEP_ALIVE>>8;/*保持连接字段，占用两个字节*/
	var_header[9] = KEEP_ALIVE&0xFF;

	remainLen = sizeof(var_header)+payload_len; /*剩余长度，也就是可变报头加上负载的长度*/

	if (remainLen > 127) {
	    fixedHeaderSize++;// add an additional byte for Remaining Length          
	}
   
	fixed_header = (uint8 *)malloc(fixedHeaderSize); /*固定报头*/
	// Message Type
	*fixed_header = MQTT_MSG_CONNECT;/*报文类型,connect*/

	
	if (remainLen <= 127) {// Remaining Length,剩余长度计算，可变长编码
	    *(fixed_header+1) = remainLen;
	} else {
	    // first byte is remainder (mod) of 128, then set the MSB to indicate more bytes
	    *(fixed_header+1) = remainLen % 128;
	    *(fixed_header+1) = *(fixed_header+1) | 0x80;
	    // second byte is number of 128s
	    *(fixed_header+2) = remainLen / 128;
	}

	packet_length = fixedHeaderSize+sizeof(var_header)+payload_len;/*固定报头+可变报头+负载长度*/
	packet = (uint8 *)malloc(packet_length);/*分配内存*/
	memset(packet, 0, packet_length);
	memcpy(packet, fixed_header, fixedHeaderSize);/*填充固定报头*/
	free(fixed_header);	
	offset += fixedHeaderSize;
	memcpy(packet+offset, var_header, sizeof(var_header));/*填充可变报头*/
	offset += sizeof(var_header);
	
	packet[offset++] = clientidlen>>8;// Client ID - UTF encoded,填充clientid长度+clientid
	packet[offset++] = clientidlen&0xFF;
	memcpy(packet+offset, client_id, clientidlen);
	offset += clientidlen;

	if(usernamelen) {// Username - UTF encoded,填充用户名+用户名长度
		packet[offset++] = usernamelen>>8;
		packet[offset++] = usernamelen&0xFF;
		memcpy(packet+offset, user_name, usernamelen);
		offset += usernamelen;
	}

	if(passwordlen) {// Password - UTF encoded,填充用户密码+用户名密码长度
		packet[offset++] = passwordlen>>8;
		packet[offset++] = passwordlen&0xFF;
		memcpy(packet+offset, passwd, passwordlen);
		offset += passwordlen;
	}
	// Send the packet
    if (client_send(sockfd,packet, packet_length) < 0){
         free(packet);
        return -1;
    }
    free(packet);
	return 1;
}

16.3.2 CONNACK-确认连接请求

​ 服务端发送 CONNACK 报文响应从客户端收到的 CONNECT 报文。服务端发送给客户端的第一个报文必须是 CONNACK。

16.3.2.1 固定报头

bit 7 6 5 4 3 2 1 0
Byte1 MQTT 控制报文类型 (2) Reserved 保留位
0 0 1 0 0 0 0 0
Byte2 剩余长度
0 0 0 0 0 0 1 0

​ 剩余长度字段表示可变报头的长度。 对于 CONNACK 报文这个值等于 2。

16.3.2.2 可变报头

描述 7 6 5 4 3 2 1 0
连接确认标记 保留位 SP1
Byte1 0 0 0 0 0 0 0 X
连接返回码
Byte2 x x x x x x x x
**Byte1,Bit0连接确认标志**

位 7-1 是保留位且必须设置为 0,

对于bit0,如果服务端收到一个 CleanSession 为 0 的连接， 当前会话标志的值取决于服务端是否已经保存了 ClientId对应客户端的会话状态。 如果服务端已经保存了会话状态， 它必须将 CONNACK 报文中的当前会话标志设置为 1 。 如果服务端没有已保存的会话状态， 它必须将 CONNACK 报文中的当前会话设置为 0。 还需要将 CONNACK 报文中的返回码设置为 0。

连接返回码

如果服务端发送了一个包含非零返回码的 CONNACK 报文， 那么它必须关
闭网络连接。

值 返回码响应 描述
0 0x00 连接已被服务端接受
1 0x01 服务端不支持客户端请求的协议版本
2 0x02 客户端标识符是正确的 UTF-8 编码， 但服务 端不允许使用
3 0x03 网络连接已建立， 但 MQTT 服务不可用
4 0x04 用户名或密码的数据格式无效
5 0x05 客户端未被授权连接到此服务器
6-255 保留

CONNACK没有有效载荷。

16.3.2.3 CONNACK报文wireshark抓包分析

image
image1314×416 117 KB

16.3.2.4 c语言构造connect ack报文

void mqtt_connect_ack(int sockfd)
{
	uint8_t cmd[]={ 0x20/*报文类型*/, 0x02/*剩余长度*/ ,0x00,0x00/*最后两个字节可变报头表示返回状态码*/ };
	send_msg(sockfd,cmd,sizeof(cmd));
	socket_record_t *socket_record = look_up_by_sokfd(sockfd);
	if(socket_record==NULL){
		return;
	}
	socket_record->is_connect=0x01;	
}

16.3.3 PUBLISH-发布消息

PUBLISH 控制报文是指从客户端向服务端或者服务端向客户端传输一个应用消息。

image
image762×591 62.4 KB

16.3.3.1 固定报头

bit 7 6 5 4 3 2 1 0
Byte 1 MQTT报文类型（3） dup Qos等级 RETAIN
0 0 1 1 x x x x
Byte2 剩余长度
**Bit3 dup**

如果 DUP 标志被设置为 0， 表示这是客户端或服务端第一次请求发送这个 PUBLISH 报文。 如果 DUP 标志被设置为 1，表示这可能是一个早前报文请求的重发。客户端或服务端请求重发一个 PUBLISH 报文时， 必须将 DUP 标志设置为 1.。 对于 QoS0 的消息， DUP 标志必须设置为 0。

Bit1和bit2 qos等级

Qos值 bit2 bit1 描述
0 0 0 最多分发一次
1 0 1 至少分发一次
2 1 0 只分发一次
- 1 1 保留不使用

qos由发送端决定，发送端发送什么qos的消息，接收端就回复什么qos的消息。

image
image1287×533 41.2 KB

Bit0 保留标记位

一般设置为0。

剩余长度

等于可变报头的长度加上有效载荷的长度。

可变报头

可变报头按顺序包含主题名和标识符。主题，用于识别有效载荷数据应该被发布到哪一个信息通道，标识符，只有当 QoS 等级是 1 或 2 时，报文标识符（ Packet Identifier） 字段才能出现在 PUBLISH 报文中。

16.3.3.2 抓包分析PUBLISH报文

image
image1472×342 28.7 KB

16.3.3.3 构造publish 报文

int mqtt_publish_with_qos(int sockfd,const char* topic, const char* msg, uint16 msgl, uint8 retain, uint8 qos, uint16* message_id) 
{
    socket_record_t *socket_record = look_up_by_sokfd(sockfd);
    if(NULL == socket_record){
        return -1;
    }
    DEBUG_INFO("sockfd:%d",socket_record->sockfd);
	uint16 topiclen = strlen(topic);
	uint16 msglen = msgl;
	uint8 *var_header = NULL; // Topic size (2 bytes), utf-encoded topic
	uint8 *fixed_header = NULL;
	uint8 fixedHeaderSize = 0,var_headerSize = 0; // Default size = one byte Message Type + one byte Remaining Length
	uint16 remainLen = 0;
	uint8 *packet = NULL;
	uint16 packet_length = 0;

	uint8 qos_flag = MQTT_QOS0_FLAG; /*qos标记*/
	uint8 qos_size = 0; // No QoS included
	if(qos == 1) {
		qos_size = 2; // 2 bytes for QoS
		qos_flag = MQTT_QOS1_FLAG;
	}
	else if(qos == 2) {
		qos_size = 2; // 2 bytes for QoS
		qos_flag = MQTT_QOS2_FLAG;
	}

	// Variable header
	var_headerSize = topiclen/*主题内容*/+2/*主题长度占用两字节*/+qos_size/*标识符*/;
	var_header = (uint8 *)malloc(var_headerSize);
	memset(var_header, 0, var_headerSize);
	*var_header = topiclen>>8;
	*(var_header+1) = topiclen&0xFF;
	memcpy(var_header+2, topic, topiclen);
	if(qos_size) {//qos1和qos2的报文需要填充标识符,有点像tcp的seq
        socket_record->publish_seq++;
        if(socket_record->publish_seq == 0){
            //unsigned short 表示范围0~65535,标识符必须是非零整数
            socket_record->publish_seq = 1;
        }
		var_header[topiclen+2] = (socket_record->publish_seq & 0xff00)>>8;
		var_header[topiclen+3] = socket_record->publish_seq & 0x00ff;
		if(message_id) { 
			*message_id = socket_record->publish_seq;
		}
	}

	fixedHeaderSize = 2;    // Default size = one byte Message Type + one byte Remaining Length
	remainLen = var_headerSize+msglen;
	if (remainLen > 127) {/*剩余长度*/
		fixedHeaderSize++;          // add an additional byte for Remaining Length
	}
	fixed_header = (uint8 *)malloc(fixedHeaderSize);/*固定报头+剩余长度*/
    
	// Message Type, DUP flag, QoS level, Retain
	*fixed_header = MQTT_MSG_PUBLISH | qos_flag;/*报文类型和qos标记*/
	if(retain) {
		*fixed_header  |= MQTT_RETAIN_FLAG;/*是否保留*/
	}
	// Remaining Length，剩余长度
	if (remainLen <= 127) {
	   *(fixed_header+1) = remainLen;
	} else {
	   // first byte is remainder (mod) of 128, then set the MSB to indicate more bytes
	   *(fixed_header+1) = remainLen % 128;
	   *(fixed_header+1) = *(fixed_header+1) | 0x80;
	   // second byte is number of 128s
	   *(fixed_header+2) = remainLen / 128;
	}

	packet_length = fixedHeaderSize+var_headerSize+msglen;/*固定报头+可变报头+负载长度*/
	packet = (uint8 *)malloc(packet_length);
	memset(packet, 0, packet_length);
	memcpy(packet, fixed_header, fixedHeaderSize);/*填充固定报头*/
	memcpy(packet+fixedHeaderSize, var_header, var_headerSize);/*填充可变报头*/
	memcpy(packet+fixedHeaderSize+var_headerSize, msg, msglen);/*负载*/
	free(var_header);
	free(fixed_header);
    send_msg(sockfd,packet , packet_length);
	free(packet);
	return 1;
}