MQTT协议分析与c语言实现(下)_mqtt源代码解析c

MQTT协议分析与c语言实现(下)

一、MQTT报文分析

1.1 CONNECT-连接服务端

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

在一个网络连接上，客户端只能发送一次 CONNECT 报文。服务端必须将客户端发送的第二个 CONNECT报文当作协议违规处理并断开客户端的连接。
有效载荷包含一个或多个编码的字段。 包括客户端的唯一标识符， Will 主题， Will 消息， 用户名和密码。 除了客户端标识之外， 其它的字段都是可选的， 基于标志位来决定可变报头中是否需要包含这些字段。

1.1.1 connect固定报头

1.1.2 协议名字节组成

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

1.1.2 协议级别

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

1.1.3 连接标记

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 。

1.1.4 保持连接

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

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

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

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

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

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

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

1.1.5 客户端标识符

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

1.1.6 遗嘱主题和遗嘱消息

如果遗嘱标志被设置为 1， 有效载荷的下一个字段是遗嘱主题（Will Topic） 。 遗嘱主题必须是 UTF-8 编码字符串。
如果遗嘱标志被设置为 1， 有效载荷的下一个字段是遗嘱消息。 遗嘱消息定义了将被发布到遗嘱主题的应用消息。

1.1.7 用户名和密码

如果用户名（ User Name） 标志被设置为 1， 有效载荷的下一个字段就是它。 用户名必须是定义的UTF-8 编码字符串。服务端可以将它用于身份验证和授权。
如果密码（ Password） 标志被设置为 1， 有效载荷的下一个字段就是它。密码字段包含一个两字节的长度字段， 长度表示二进制数据的字节数（ 不包含长度字段本身占用的两个字节），后面跟着 0 到 65535 字节的二进制数据。

1.1.8 wirshark抓包分析connect报文

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

1.1.9 c代码封装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;

	// 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+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);
	// Client ID - UTF encoded
	packet[offset++] = clientidlen>>8;
	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 (send_msg(sockfd,packet, packet_length) < 0){
         free(packet);
        return -1;
    }
    free(packet);
	return 1;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

1.2 CONNACK-确认连接请求

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

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

1.2.1 可变报头

Byte1,Bit0连接确认标志，位 7-1 是保留位且必须设置为 0,
对于bit0,如果服务端收到一个 CleanSession 为 0 的连接， 当前会话标志的值取决于服务端是否已经保存了 ClientId对应客户端的会话状态。 如果服务端已经保存了会话状态， 它必须将 CONNACK 报文中的当前会话标志设置为 1 。 如果服务端没有已保存的会话状态， 它必须将 CONNACK 报文中的当前会话设置为 0。 还需要将 CONNACK 报文中的返回码设置为 0。
连接返回码
如果服务端发送了一个包含非零返回码的 CONNACK 报文， 那么它必须关
闭网络连接。

CONNACK没有有效载荷。

1.2.2 CONNACK报文wireshark抓包分析

1.2.3 c代码封装CONNACK报文

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;	
}
1
2
3
4
5
6
7
8
9
10

1.3 PUBLISH-发布消息

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

1.3.1 固定报头

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

Bit1和bit2 qos等级

qos由发送端决定，发送端发送什么qos的消息，接收端就回复什么qos的消息。
不同qos等级mqtt报文交互流程

Bit0 保留标记位
一般设置为0。

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

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

1.3.2 抓包分析PUBLISH报文

1.3.3 c代码封装PUBLISH报文（qos0,qos1,qos2）

int mqtt_publish(int sockfd,const char* topic, const char* msg, uint16 msglen, uint8 retain) 
{
	return mqtt_publish_with_qos(sockfd,topic, msg, msglen, retain, QOS_VALUE2, NULL);
}

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 = strlen(msg);
	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;
	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) {
        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;
		}
	}

	// Fixed header
	// the remaining length is one byte for messages up to 127 bytes, then two bytes after that
	// actually, it can be up to 4 bytes but I'm making the assumption the embedded device will only
	// need up to two bytes of length (handles up to 16,383 (almost 16k) sized message)
	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;
	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);
	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;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

1.4 PUBREC-发布收到

PUBREC 报文是对 QoS 等级 2 的 PUBLISH 报文的响应。它是 QoS 2 等级协议交换的第二个报文。

1.4.1 固定报头

剩余长度
表示可变报头的长度。 对 PUBREC 报文它的值等于 2。
可变报头

有效载荷
PUBREC 报文没有有效载荷。

1.4.2 PUBREC抓包报文

1.4.3 c代码封装PUBREC报文

//如果是PUBREC报文，head_type=0x50
void mqtt_qos2_pubrec(int   sockfd , unsigned char *data,unsigned char head_type)
{
    uint16 msg_id = mqtt_parse_msg_id(data);
	unsigned char qos2_pubrec_respon[]={head_type ,0x02 , (msg_id & 0xff00)>>8 , msg_id & 0x00ff};
	send_msg(sockfd,qos2_pubrec_respon,sizeof(qos2_pubrec_respon));
}
1
2
3
4
5
6
7

1.5 PUBREL-发布释放

PUBREL 报文是对 PUBREC 报文的响应。 它是 QoS 2 等级协议交换的第三个报文。

1.5.1 固定报头

剩余长度
表示可变报头的长度。 对 PUBREL 报文它的值等于 2。
可变报头

有效载荷
PUBREL 报文没有有效载荷。

1.5.2 PUBREL抓包报文

1.5.3 c语言封装PUBREL报文

//head_type=0x62
void mqtt_qos2_pubrel(int   sockfd , unsigned char *data,unsigned char head_type)
{
    uint16 msg_id = mqtt_parse_msg_id(data);
	unsigned char qos2_pubrel_respon[]={head_type ,0x02 , (msg_id & 0xff00)>>8 , msg_id & 0x00ff};
	send_msg(sockfd,qos2_pubrel_respon,sizeof(qos2_pubrel_respon));
}
1
2
3
4
5
6
7

1.6 PUBCOMP-发布完成

PUBCOMP 报文是对 PUBREL 报文的响应。 它是 QoS 2 等级协议交换的第四个也是最后一个报文。

1.6.1固定报头

剩余长度
表示可变报头的长度。 对 PUBCOMP 报文它的值等于 2。
可变报头

有效载荷
PUBCOMP 报文没有有效载荷。

1.6.2 PUBCOMP抓包报文

1.6.3 c语言封装PUBCOMP报文

//head_type=0x70
void mqtt_qos2_pubcomp(int   sockfd , unsigned char *data,unsigned char head_type)
{
    uint16 msg_id = mqtt_parse_msg_id(data);
	unsigned char qos2_pubcomp_respon[]={head_type ,0x02 , (msg_id & 0xff00)>>8 , msg_id & 0x00ff};
	send_msg(sockfd,qos2_pubcomp_respon,sizeof(qos2_pubcomp_respon));
}
1
2
3
4
5
6
7

1.7 PINGREQ-心跳请求

客户端发送 PINGREQ 报文给服务端的。用于：
a) 在没有任何其它控制报文从客户端发给服务的时，告知服务端客户端还活着。
b) 请求服务端发送 响应确认它还活着。
c) 使用网络以确认网络连接没有断开。

1.7.1 固定报头

可变报头
报文没有可变报头。
有效载荷
PINGREQ 报文没有有效载荷。

1.7.2 PINGREQ 抓包报文

1.7.3 c代码封装PINGREQ报文

int mqtt_ping(int sockfd)
{
	uint8 packet[] = {
		MQTT_MSG_PINGREQ, // Message Type, DUP flag, QoS level, Retain
		0x00 // Remaining length
	};
	int ret = send_msg(sockfd,packet, sizeof(packet));
    //DEBUG_INFO("ret=%d",ret);
	return ret;
}
1
2
3
4
5
6
7
8
9
10

1.8 PINGRESP – 心跳响应

服务端发送 PINGRESP 报文响应客户端的 PINGREQ 报文。 表示服务端还活着。

1.8.1 固定报头

可变报头
报文没有可变报头。
有效载荷
PINGRESP 报文没有有效载荷。

1.8.2 PINGRESP 抓包报文

1.8.3 c语言封装PINGRESP报文

void mqtt_ping_req_reply(int sockfd)
{
    uint8_t cmd[]={0xd0, 0x00};
    send_msg(sockfd,cmd,sizeof(cmd));
}
1
2
3
4
5

1.9 DISCONNECT –断开连接

DISCONNECT 报文是客户端发给服务端的最后一个控制报文。表示客户端正常断开连接。

1.9.1 固定报头

可变报头
DISCONNECT报文没有可变报头。
有效载荷
DISCONNECT 报文没有有效载荷。

1.9.2 DISCONNECT 抓包报文

1.9.3 c语言封装DISCONNECT报文

int mqtt_disconnect(int sockfd) 
{
	uint8 packet[] = {
		MQTT_MSG_DISCONNECT, // Message Type, DUP flag, QoS level, Retain
		0x00 // Remaining length
	};
	int ret = client_send(sockfd,packet, sizeof(packet));
    DEBUG_INFO("ret=%d",ret);
	return ret;
}
1
2
3
4
5
6
7
8
9
10

1.10 SUBSCRIBE-订阅主题

客户端向服务端发送 SUBSCRIBE 报文用于创建一个或多个订阅。 每个订阅注册客户端关心的一个或多个主题。 为了将应用消息转发给与那些订阅匹配的主题， 服务端发送 PUBLISH 报文给客户端。 SUBSCRIBE报文也（为每个订阅）指定了最大的 QoS 等级， 服务端根据这个发送应用消息给客户端。

1.10.1 固定报头

SUBSCRIBE 控制报固定报头的第 3,2,1,0 位是保留位， 必须分别设置为 0,0,1,0,服务端必须将其它的任何值都当做是不合法的并关闭网络连接。

剩余长度字段
等于可变报头的长度（ 2 字节） 加上有效载荷的长度。
可变报头

服务端收到客户端发送的一个 SUBSCRIBE 报文时， 必须使用 SUBACK 报文响应。SUBACK 报文必须和等待确认的 SUBSCRIBE 报文有相同的报文标识符。

有效载荷

QoS 不等于 0,1 或 2， 服务端必须认为 SUBSCRIBE 报文是不合法的并关闭网络连接。

1.10.2 SUBSCRIBE报文抓包

1.10.3 c代码封装SUBSCRIBE报文

static uint16 su_seq = 1;
int mqtt_subscribe_theme(int sockfd,char *Theme , uint8_t Qos)
{
	su_seq++;
	if(su_seq == 0){
		su_seq = 1;
	}
	uint16_t MessageId = su_seq;
	uint8_t cmd[1024]={0};
    //报文标示符长度2 + 主题长度位占用2字节+主题长度+qos标识
	int data_length = 2+2+strlen(Theme)+1; 
	int playload_len = strlen(Theme);
	uint8_t len_byte[4] ={0x00 , 0x00 ,0x00 ,0x00};
	uint8_t byte_num = length_trans_byte_form(data_length , len_byte);
	cmd[0] = 0x82;
	memcpy(&cmd[1] , len_byte , byte_num);

	cmd[1+byte_num]=(MessageId & 0xff00) >> 8 ;
	cmd[1+byte_num+1] = MessageId & 0x00ff;		
	cmd[1+byte_num+1+1] = (playload_len & 0xff00) >> 8;
	cmd[1+byte_num+1+1+1] = playload_len & 0x00ff;
	memcpy(&cmd[1+byte_num+1+1+1+1] , Theme , playload_len);
	cmd[1+byte_num+1+1+1+1+playload_len] = Qos;
	client_send(sockfd,cmd, 1+byte_num+1+1+1+1+playload_len+1);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

1.11 SUBACK –订阅确认

服务端发送 SUBACK 报文给客户端，用于确认它已收到并且正在处理 SUBSCRIBE 报文。

1.11.1固定报头

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

可变报头包含等待确认的 SUBSCRIBE 报文的报文标识符。

1.11.2 SUBACK抓包报文

1.11.3 c代码封装SUBACK

void mqtt_subscribe_ack(int sockfd,const uint8 *buf)
{
    uint16 msg_id = mqtt_parse_msg_id(buf);
    uint8 qos = MQTTParseMessageQos(buf);
    uint8 cmd[]={0x90,0x03/*剩余长度*/, (msg_id & 0xff00) >> 8, msg_id & 0x00ff,qos};
    send_msg(sockfd,cmd,sizeof(cmd));
}
1
2
3
4
5
6
7

1.12 UNSUBSCRIBE –取消订阅

客户端发送 UNSUBSCRIBE 报文给服务端， 用于取消订阅主题。

1.12.1 固定报头

UNSUBSCRIBE 报文固定报头的第 3,2,1,0 位是保留位且必须分别设置为 0,0,1,0。 服务端必须认为任何其它的值都是不合法的并关闭网络连接。
剩余长度字段
等于可变报头的长度，加上有效载荷的长度。
可变报头

可变报头包含一个报文标识符。
有效载荷

UNSUBSCRIBE 报文的有效载荷必须至少包含一个消息过滤器。 没有有效载荷的 UNSUBSCRIBE 报文是违反协议的。

1.12.2 UNSUBSCRIBE抓包报文

1.12.3 c代码封装UNSUBSCRIBE报文

static uint16 un_seq = 1;
int mqtt_unsubscribe_theme(int sockfd,const char* topic)
{
	un_seq++;
	if(un_seq == 0){
		un_seq = 1;
	}
    uint16_t MessageId = un_seq;
	uint8_t cmd[1024]={0};
    //报文标示符长度2 + 主题长度位占用2字节+主题长度+qos标识
	int data_length = 2+2+strlen(topic)+1; 
	int playload_len = strlen(topic);
	uint8_t len_byte[4] ={0x00 , 0x00 ,0x00 ,0x00};
	uint8_t byte_num = length_trans_byte_form(data_length , len_byte);
	cmd[0] = 0xa2;
	memcpy(&cmd[1] , len_byte , byte_num);
    
	cmd[1+byte_num]=(MessageId & 0xff00) >> 8 ;
	cmd[1+byte_num+1] = MessageId & 0x00ff;		
	cmd[1+byte_num+1+1] = (playload_len & 0xff00) >> 8;
	cmd[1+byte_num+1+1+1] = playload_len & 0x00ff;
	memcpy(&cmd[1+byte_num+1+1+1+1] , topic , playload_len);
	client_send(sockfd,cmd,1+byte_num+1+1+1+1+playload_len+1);
	return 1;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

1.13 UNSUBACK –取消订阅确认

服务端发送 UNSUBACK 报文给客户端用于确认收到 UNSUBSCRIBE 报文。

1.13.1 固定报头

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

可变报头包含等待确认的 UNSUBSCRIBE 报文的报文标识符。

1.13.2 UNSUBSCRIBE ACK抓包报文

1.13.3 c代码封装UNSUBSCRIBE ACK报文

void mqtt_unsubscribe_ack(int sockfd,const uint8 *buf)
{
    uint16 msg_id = mqtt_parse_msg_id(buf);
    uint8 cmd[]={0xb0,0x02/*剩余长度*/,(msg_id & 0xff00) >> 8, msg_id & 0x00ff};
    send_msg(sockfd,cmd,sizeof(cmd));
}
1
2
3
4
5
6

1.4 mqtt客户端与服务端完整代码

https://download.csdn.net/download/ZZEZZEZY/12405817

1.5 参考

文档《MQTT-3.1.1-CN.pdf》
https://download.csdn.net/download/ZZEZZEZY/11986143
参考代码：https://github.com/menudoproblema/libemqtt.git