C语言实现MQTT协议(三)源代码介绍及连接阿里云_mqtt源码

接下来介绍C语言实现MQTT的源代码文件。

数据发送缓冲区

static char MQTTSendBuff[MQTT_BUFF_SIZE] = { 0 };

定义一个数据发送缓冲区，用来存储需要发送的数据，其中宏定义MQTT_BUFF_SIZE在头文件中已定义，因为该缓冲区只在该C文件中使用，所以可以加上static关键字。

接口

发送数据demo

这是客户端向服务端发送数据的接口，需要我们根据自己平台来实现该函数。
这里我给出2个demo。

demo1
将数据发送到控制台窗口中

/** \brief 发送数据-----接口  示例
 *
 * \param   data 指向需要发送的数据的指针
 * \param   length 数据的长度 单位字节
 * \return  返回1则发送成功，返回0则发送失败
 *
 */
static int Demo_MQTTSendDataToServer(const char* data, unsigned int length)
{
	/*
		这里使用了C语言库函数实现该接口
		如果要自己实现请把下面的代码删除，
		并将lMQTT.h文件中头文件
		#include <stdio.h>
		#include <time.h>
		去除
	*/
	unsigned char temp = 0;
	while (length--)
	{
		temp = *data++;
		printf("%02X ", temp);
	}
	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

demo2
通过TCP/IP发送数据。

/** \brief 发送数据-----接口  示例
 *
 * \param   data 指向需要发送的数据的指针
 * \param   length 数据的长度 单位字节
 * \return  返回1则发送成功，返回0则发送失败
 *
 */
static int Demo_MQTTSendDataToServer(const char* data, unsigned int length)
{
	/*
		这里使用了C语言库函数实现该接口
		如果要自己实现请把下面的代码删除，
		并将lMQTT.h文件中头文件
		#include <time.h>
		#include <windows.h>
		#pragma comment (lib, "ws2_32.lib")  //加载 ws2_32.dll
		去除
	*/
	if (send(sock, data, length, 0) > 0)
	{
		return 1;
	}
	return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

毫秒级延时demo

实现毫秒级延时

/** \brief 延时毫秒函数-----接口  示例
 *
 * \param   ms 毫秒数
 * \return  无
 *
 */
static void Demo_MQTTDelayms(unsigned int ms)
{
	/*
	这里使用了C语言库函数实现该接口
	如果要自己实现请把下面的代码删除，
	并将lMQTT.h文件中头文件
	#include <stdio.h>
	#include <time.h>
	去除
	*/
	clock_t start = clock();
	while (clock() - start < ms);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

接收数据demo

这是客户端接收到服务端的数据所要调用的函数，请根据自己的需要更改完善函数，比如我们接收到一个开关量的数据a，这时候就根据需求添加代码。

/** \brief 从服务器接收到数据-----接口  示例
 *          (未写QoS = 1/2的响应代码和收到来自服务器的数据并解析数据，这个需要结合自己的情景需要修改代码)
 *
 * \param   data 指向接收到的数据的指针
 * \param   length 数据的长度 单位字节
 * \return  无
 *
 */
void Demo_MQTTReceiveDadaFromServer(unsigned char* data, unsigned int length)
{
	switch (*data)
	{
		/* 确认连接请求 */
	case MQTT_CONNACK:
		//避免野指针
		if (mqtt.returnData)
		{
			//获取当前会话标志
			((MQTTConnACKStruct_t*)mqtt.returnData)->sp = *(data + 2);
			//获取连接返回码
			((MQTTConnACKStruct_t*)mqtt.returnData)->code = *(data + 3);
		}
		mqtt.resultCode = ~MQTT_RESULT_CODE_INIT;
		break;

		/* 确认发布消息请求(未写QoS = 1/2的响应代码) */
	case MQTT_PUBACK:
	case MQTT_PUBREC:
	case MQTT_PUBREL:
	case MQTT_PUBCOMP:
		mqtt.resultCode = ~MQTT_RESULT_CODE_INIT;
		break;

		/* 订阅确认 */
	case MQTT_SUBACK:
		//避免野指针
		if (mqtt.returnData)
		{
			((MQTTSubACKStruct_t*)mqtt.returnData)->messageID = *(data + 2) * 256 + *(data + 3);
			((MQTTSubACKStruct_t*)mqtt.returnData)->code = *(data + 4);
		}
		mqtt.resultCode = ~MQTT_RESULT_CODE_INIT;
		break;

		/* 取消订阅确认 */
	case MQTT_UNSUBACK:
		//避免野指针
		if (mqtt.returnData)
		{
			((MQTTUnsubACKStruct_t*)mqtt.returnData)->messageID = *(data + 2) * 256 + *(data + 3);
		}
		mqtt.resultCode = ~MQTT_RESULT_CODE_INIT;
		break;

		/* 心跳响应 */
	case MQTT_PINGRESP:
		mqtt.resultCode = ~MQTT_RESULT_CODE_INIT;
		break;

		/*
			在这里添加自己的业务代码
		*/
	default:
		break;
	}

	//清空缓冲区
	memset((void*)data, 0, length);
}
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

注意一下：
函数参数data是unsigned char *类型
但是MQTT_SUBACK(0x90)订阅响应报文，MQTT_UNSUBACK(0xB0)取消订阅响应报文等等报文，在switch代码块里，这些枚举类型是被当成int类型，如MQTT_SUBACK(0x90)等于十进制数144，假如我们把函数参数data定义为char *类型，data 16进制数为MQTT_SUBACK(0x90)，但当我们取值时(0x90)被解释为char类型的数故 *data等于-122，故我们需要将函数参数data定义为unsigned char *类型

或者我们可以在每一个case后加一个强制类型转换，如

/* 确认连接请求 */
case (char)MQTT_CONNACK
	break;

/* 确认发布消息请求(未写QoS = 1/2的响应代码) */
case (char)MQTT_PUBACK:
case (char)MQTT_PUBREC:
case (char)MQTT_PUBREL:
case (char)MQTT_PUBCOMP:
	break;

/* 订阅确认 */
case (char)MQTT_SUBACK:
	break;

/* 取消订阅确认 */
case (char)MQTT_UNSUBACK:
	break;
	
/* 心跳响应 */
case (char)MQTT_PINGRESP:
	break;
	
default:
	break;
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

模块化函数

将数据写入到MQTT的数据发送缓冲区

这是将数据写入到MQTT的数据发送缓冲区MQTTSendBuff，其中包括了写入字符和实数。

/** \brief  将数据写入MQTT的发送缓冲区
 * \param   data 指向要写入的数据源，类型强制转换为 void* 指针。
 * \param   n 要被写入的字节数。
 * \param   dataType  说明数据源 data 的数据类型，如果是字符则传入CHAR(1)，不是则传入NUM(0)。
 * \return   无。
 */
static void MQTTSendDataToBuff(void* data, unsigned int n, DataType_t dataType)
{
	if (data == NULL)
		return;
	if (dataType == MQTT_NUM)
		data = (char*)data + n - 1;
	while (n--)
	{
		*mqtt.sendBuffPointNow++ = *(char*)data;
		data = (char*)data + dataType;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

这里解释一下数量类型类型DataType_t 。

/* 数据类型枚举 */
typedef enum
{
	MQTT_NUM = -1,				//数据是实数
	MQTT_CHAR = 1,				//数据是字符
}DataType_t;
1
2
3
4
5
6

为什么实数的枚举值是-1，而字符的枚举值是1。
首先mqtt.sendBuffPointNow指针指向的是MQTT数据发送缓冲区的下一个要写入的地址。
在小端模式下，发送字符串的时，传入的字符指针指向的是第一个字符的地址，如字符串char *str = Hello; str指向的是字符H的地址，当我们调用这个函数的时候情况就如下图，每写入一个字符后，mqtt.sendBuffPointNow会指向下一个地址，str也就是函数中变量data也会自增，指向下一个要写入的字符的地址，与我们所期望的情况是一致。

但是当我们发送实数的时候，在小端模式下，如short s = 0x1122; &s指向的是低字节的数据的地址。
变量s在内存的存储情况:

假如我们不加调整。就会出现下图所示的情况。

这与我们期望的情况正好相反，因为MQTT协议是先发高字节，高字节数据在前。
所以我们需要将函数中变量data偏移至指向最高高字节数据的地址，然后每写入一个字节后data自减，这样才能符号我们期望的情况。

编码剩余长度

根据剩余长度规则，编写了编码函数。
其中定义一个数组先暂存已经编码的剩余长度，然后反向将数组写入到MQTT数据发送缓冲区，因为编码剩余长度先得到是低字节的数据，而我们是高字节在前发送出去，故需要反转。

/** \brief  编码剩余长度并向发送缓冲区写入报文的类型
 *
 * \param   无
 * \return  无
 *
 */
static void codeRemainLengthAndSendMessageType(void)
{
	char temp[4] = { 0 };
	unsigned char length = 0;
	char encodedByte = 0;
	unsigned int X = 0;

	//计算剩余长度
	X = mqtt.sendBuffPointNow - mqtt.sendBuff - 1;
	//编码
	do
	{
		encodedByte = X % 128;
		X = X / 128;
		if (X > 0)
		{
			encodedByte = encodedByte | 128;
		}
		//得到已编码的剩余长度
		temp[length] = encodedByte;
		//记录编码的剩余长度占用的字节数
		length++;
	} while (X > 0);

	//反转
	while (length--)
	{
		*mqtt.sendBuff = temp[length];
		mqtt.sendBuff--;
	}

	//写入报文的类型
	*mqtt.sendBuff = mqtt.messageType;
}
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

解码剩余长度

/** \brief  解码剩余长度
 *
 * \param   data 指向已编码的剩余长度数组的首个元素的指针
 * \return  无
 *
 */
void decodeRemainLength(const char* data)
{
	unsigned int multiplier = 1;
	unsigned int value = 0;
	unsigned char encodedByte = 0;
	do {
		encodedByte = *data++;
		value += (encodedByte & 127) * multiplier;
		multiplier *= 128;
		if (multiplier > 128 * 128 * 128) {
			// throw Error(Malformed Remaining Length)
			// error
			return;
		}
	} while ((encodedByte & 128) != 0);

	//得到已解码的剩余长度
	mqtt.remainLength = value;
}
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

初始化函数

主要是初始化MQTT的结构体变量mqtt，把数据发送缓冲区偏移5个字节来存储报头和剩余长度，因为剩余长度需要根据可变报头和有效载荷才能计算。

/** \brief  MQTT初始化
 *
 * \param   无
 * \return  无
 *
 */
void MQTTInit(void)
{
	mqtt.messageType = 0;
	mqtt.sendBuff = MQTTSendBuff + 4;
	mqtt.sendBuffPointNow = MQTTSendBuff + 5;
	mqtt.resultCode = MQTT_RESULT_CODE_INIT;
	mqtt.returnData = NULL;

	if (mqtt.MQTTSendDataToServer == NULL)
	{
		mqtt.MQTTSendDataToServer = Demo_MQTTSendDataToServer;
	}
	if (mqtt.MQTTDelayms == NULL)
	{
		mqtt.MQTTDelayms = Demo_MQTTDelayms;
	}
	if (mqtt.MQTTReceiveDadaFromServer == NULL)
	{
		mqtt.MQTTReceiveDadaFromServer = Demo_MQTTReceiveDadaFromServer;
	}
}
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

功能函数

这里我只介绍某些函数，比如CONNECT连接服务端函数。

CONNECT连接服务端函数

连接服务端函数，先初始化结构体，存储指向CONNACK结构体变量的指针的值，根据写入固定报头，可变报头，有效载荷，之后等待服务端响应，在客户端接收到服务端函数MQTTReceiveData中判断连接状态，如果没有等待超时，则返回0。我们可以根据变量s2来判断连接状态并做出响应动作比如错误处理或者重新连接。

/** \brief  MQTT连接报文
 *
 * \param 	s1 指向连接报文结构体的指针
 * \param 	s2 指向确认连接请求结构体的指针
 * \return 	返回 0 连接成功
 *			返回 1 连接超时，等待连接的时间已经超过最大超时等待时间(MQTT_MAX_TIMES_OUT)
 *
 */
char MQTTConnect(MQTTConnectStruct_t* s1, MQTTConnACKStruct_t* s2)
{
	unsigned short temp = 0;

	MQTTInit();
	mqtt.returnData = s2;

	//报文的类型
	mqtt.messageType = MQTT_CONNECT;

	/* 可变报头 */
	//协议名
	temp = 0x0004;
	MQTTSendDataToBuff(&temp, 2, MQTT_NUM);
	MQTTSendDataToBuff("MQTT", 4, MQTT_CHAR);
	//协议级别
	MQTTSendDataToBuff(&temp, 1, MQTT_NUM);
	//连接标志
	MQTTSendDataToBuff(&s1->connectFlag, 1, MQTT_NUM);
	//保持连接时间
	MQTTSendDataToBuff(&s1->keepAliveTime, 2, MQTT_NUM);

	/* 有效载荷 */
	//客户端标识长度
	temp = strlen(s1->clientID);
	MQTTSendDataToBuff(&temp, 2, MQTT_NUM);
	//客户端标识
	MQTTSendDataToBuff((void*)s1->clientID, temp, MQTT_CHAR);
	//用户名长度
	temp = strlen(s1->userName);
	MQTTSendDataToBuff(&temp, 2, MQTT_NUM);
	//用户名
	MQTTSendDataToBuff((void*)s1->userName, temp, MQTT_CHAR);
	//密码长度
	temp = strlen(s1->password);
	MQTTSendDataToBuff(&temp, 2, MQTT_NUM);
	//密码
	MQTTSendDataToBuff((void*)s1->password, temp, MQTT_CHAR);

	/* 编码剩余长度 */
	codeRemainLengthAndSendMessageType();

	/* 计算需要发送的数据的长度并将数据发送到服务器 */
	mqtt.MQTTSendDataToServer(mqtt.sendBuff, mqtt.sendBuffPointNow - mqtt.sendBuff);

	/* 等待服务器响应 */
	temp = 0;
	while (mqtt.resultCode == MQTT_RESULT_CODE_INIT)
	{
		mqtt.MQTTDelayms(1);
		temp++;
		if (temp > MQTT_MAX_TIMES_OUT)
		{
			return 1;//超时返回
		}
	}

	return 0;
}
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

PUBLISH发布消息函数

客户端发布消息至服务端。调用模块化函数来编写，记得初始化MQTT结构体变量。

/** \brief  MQTT发布消息(可结合自己的情景修改)
 *
 * \param 	s1 指向发布消息结构体的指针
 * \param 	s2 指向发布确认结构体的指针(如果QoS = 0，则可以将传入NULL给s2)
 * \return 	返回 0 成功
 *			返回 1 连接超时，等待连接的时间已经超过最大超时等待时间(MQTT_MAX_TIMES_OUT)
 *
 */
char MQTTPublish(MQTTPublishStruct_t* s1, MQTTPubACKStruct_t* s2)
{
	unsigned short temp = 0;

	MQTTInit();
	mqtt.returnData = s2;

	//报文的类型
	mqtt.messageType = MQTT_PUBLISH;
	mqtt.messageType |= s1->RETAIN;
	mqtt.messageType |= s1->QoS << 1;
	mqtt.messageType |= s1->DUP << 3;

	/* 可变报头 */
	//主题长度
	temp = strlen(s1->topic);
	MQTTSendDataToBuff(&temp, 2, MQTT_NUM);
	//主题
	MQTTSendDataToBuff((void*)s1->topic, temp, MQTT_CHAR);
	//报文标识符
	if (s1->QoS)
	{
		MQTTSendDataToBuff(&s1->messageID, 2, MQTT_NUM);
	}

	/* 有效载荷 */
	temp = strlen(s1->payload);
	MQTTSendDataToBuff((void*)s1->payload, temp, MQTT_CHAR);

	/* 编码剩余长度 */
	codeRemainLengthAndSendMessageType();

	/* 计算需要发送的数据的长度并将数据发送到服务器 */
	MQTTSendData(mqtt.sendBuff, mqtt.sendBuffPointNow - mqtt.sendBuff);

	//QoS = 0 服务器没有响应动作
	if (!s1->QoS)
		return 0;

	/* 等待服务器响应 */
	temp = 0;
	while (mqtt.resultCode == MQTT_RESULT_CODE_INIT)
	{
		mqtt.MQTTDelayms(1);
		temp++;
		if (temp > MQTT_MAX_TIMES_OUT)
		{
			return 1;//超时返回
		}
	}

	return 0;
}
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

就贴上2个函数的代码，其他的函数差不多。

连接阿里云

CONNECT连接服务端

连接标志

保活时间

查阅阿里云MQTT协议规范，由于使用阿里云平台必须使用用户名和密码登录，所以位7和位6为1，位5到位2为遗嘱标志，阿里云平台不支持，都为0，位1为清理会话功能，此处设为1，位0保持为0，所以连接标志位为0xC2。

我们设置为300秒。

有效载荷(客户端，用户名，密码)

CONNECT 报文的有效载荷（payload）包含一个或多个以长度为前缀的字段，可变报头中的标志决定是否 包含这些字段。如果包含的话，必须按这个顺序出现：客户端标识符，遗嘱主题，遗嘱消息，用户名，密码。根据标志位来确定这些字段是否出现。
连接标志为C2也就是只有用户名和密码。

根据MQTT-TCP连接通信文档，得到客户端标识符，用户名，密码格式。

假设clientId = 12345，deviceName = device， productKey = pk， timestamp = 789，signmethod=hmacsha1，deviceSecret=secret，那么使用TCP方式提交给MQTT的参数如下：

mqttclientId=12345|securemode=3,signmethod=hmacsha1,timestamp=789|
mqttUsername=device&pk
mqttPassword=hmacsha1("secret","clientId12345deviceNamedeviceproductKeypktimestamp789").toHexString(); 
1
2
3

至此，可以得到

int Connect(void)
{
	/* MQTT连接报文 */
	MQTTConnectStruct_t s1 = { 0 };
	MQTTConnACKStruct_t s2 = { 0 };
	s1.connectFlag = 0xC2;//连接标志
	s1.keepAliveTime = MQTT_KEEP_ALIVE;//保活时间
	s1.clientID = CLIENTID;//填入你的客户端标识符
	s1.userName = USERNAME;//你的用户名标识符
	s1.password = PASSWORD;//你的密码
	return MQTTConnect(&s1, &s2);//连接阿里云
}
1
2
3
4
5
6
7
8
9
10
11
12

PUBLISH发布消息

可变报头按顺序包含主题名和报文标识符。
只有当 QoS 等级是 1 或 2 时，报文标识符（Packet Identifier） 字段才能出现在 PUBLISH 报文中，为了简单，QoS 等级为 0。
查阅阿里云帮助文档，获得主题名格式与有效载荷格式。

int Publish(void)
{
	MQTTPublishStruct_t s1 = { 0 };
	MQTTPubACKStruct_t s2 = { 0 };

	s1.DUP = 0;
	s1.QoS = 0;//QoS = 0 故没有报文标识符 与 服务端不会发送PUBLISHACK
	//一般来说QoS = 0已经够用了，再加上阿里云有 post_reply 主题就能知道发布消息是否成功
	s1.RETAIN = 0;
	//主题
	s1.topic = TOPIC;//填写你的TOPIC
	//有效载荷
	s1.payload = PAYLOAD;//填写你的有效载荷
	return MQTTPublish(&s1, &s2);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

其他的功能函数的使用方式类似，就不举例了。

demo

用Visual Studio写的一个demo
测试结果
可以看见，
Connect = 0
Publish = 0
Subscribe = 0
UnSubscribe = 0
Ping = 0
证明功能正常，测试通过。
最后这个与服务端通信时发生异常请忽略。

有需要的可以下拉到文章末尾下载demo工程。

后续

不足与改进

本次用C语言编写MQTT协议，并没有全部实现，这里我只是根据自己的需求编写，可能有考虑不到的地方，欢迎大家批评指正；后续可以添加错误代码，错误回调函数，我们可以根据不同的错误代码做出不同的解决措施。

C语言实现MQTT协议(一)协议讲解

C语言实现MQTT协议(二)头文件介绍

C语言实现MQTT协议(三)源代码介绍及连接阿里云

源代码下载链接