C语言实现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 
		#include 
		去除
	*/
	unsigned char temp = 0;
	while (length--)
	{
		temp = *data++;
		printf("%02X ", temp);
	}
	return 1;
}

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 
		#include 
		#pragma comment (lib, "ws2_32.lib")  //加载 ws2_32.dll
		去除
	*/
	if (send(sock, data, length, 0) > 0)
	{
		return 1;
	}
	return 0;
}

毫秒级延时demo

实现毫秒级延时

/** brief 延时毫秒函数-----接口  示例
 *
 * param   ms 毫秒数
 * return  无
 *
 */
static void Demo_MQTTDelayms(unsigned int ms)
{
	/*
	这里使用了C语言库函数实现该接口
	如果要自己实现请把下面的代码删除，
	并将lMQTT.h文件中头文件
	#include 
	#include 
	去除
	*/
	clock_t start = clock();
	while (clock() - start < ms);
}

接收数据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);
}

注意一下：
函数参数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;

模块化函数

将数据写入到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;
	}
}

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

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

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

mqtt.sendBuffPointNow	0	1	2	3	4
str	`H`	`e`	`l`	`l`	`o`

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

地址值	数据
0x00000000`&s`	0x22
0x00000001`((char*)&s) + 1`	0x11

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

mqtt.sendBuffPointNow	0	1
s	22	11

这与我们期望的情况正好相反，因为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;
}

解码剩余长度

/** 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;
}

初始化函数

主要是初始化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;
	}
}

功能函数

这里我只介绍某些函数，比如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;
}

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

就贴上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();

至此，可以得到

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);//连接阿里云
}

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);
}

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

demo

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

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

在这里插入图片描述

后续

不足与改进

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

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

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

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

源代码下载链接

数据发送缓冲区

接口

发送数据demo

毫秒级延时demo

接收数据demo

模块化函数

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

编码剩余长度

解码剩余长度

初始化函数

功能函数

CONNECT连接服务端函数

PUBLISH发布消息函数

连接阿里云

CONNECT连接服务端

连接标志

保活时间

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

PUBLISH发布消息

demo

后续

不足与改进

相关文章

央媒谈“文盲演员”引发饭圈甩锅大战

野生大熊猫深夜遛达 镜头前呆萌打卡

张萌谈演员到底需不需要有文化

情人节多地迎领证高峰

婚房烂尾10年业主住进毛坯房

多地首套房贷利率降至4%以下 为什么急于提前还贷

野生大熊猫深夜遛达镜头前呆萌打卡

多地首套房贷利率降至4%以下为什么急于提前还贷