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

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

数据发送缓冲区

static char MQTTSendBuff[MQTT_BUFF_SIZE] = {0};

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

接口

发送数据

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

/** \brief 发送数据-----接口
 *
 * \param   data 指向需要发送的数据的指针
 * \param   length 数据的长度 单位字节
 * \return  返回1则发送成功，返回0则发送失败
 *
 */
static char MQTTSendData(const char *data, unsigned int length)
{
    /* 请实现该函数 */
}

毫秒级延时

/** \brief 延时毫秒函数-----接口
 *
 * \param   ms 毫秒数
 * \return  无
 *
 */
static void MQTTDelayms(unsigned int ms)
{
	/* 请实现该函数 */
}

接收数据

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

/** \brief 从服务器接收到数据-----接口
 *          (未写QoS = 1/2的响应代码和收到来自服务器的数据并解析数据，这个需要结合自己的情景需要修改代码)
 *
 * \param   data 指向接收到的数据的指针
 * \param   length 数据的长度 单位字节
 * \return  无
 *
 */
void MQTTReceiveData(const 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;

	/* 请完善代码 */
	//case demo:
	//break;
    default:
        break;
    }

    //清空缓冲区
    memset((void*)data, 0, length);
}

模块化函数

将数据写入到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 == NUM)
        data = (char*)data + n - 1;
    while(n--)
    {
        *mqtt.sendBuffPointNow++ = *(char*)data;
        data = (char*)data + dataType;
    }
}

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

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

为什么实数的枚举值是-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;
}

解码剩余长度

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

功能函数

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

CONNECT连接服务端函数

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

/** \brief  MQTT连接报文
 *
 * \param s1 指向连接报文结构体的指针
 * \param s2 指向确认连接请求结构体的指针
 * \return  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, NUM);
    MQTTSendDataToBuff("MQTT", 4, CHAR);
    //协议级别
    MQTTSendDataToBuff(&temp, 1, NUM);
    //连接标志
    MQTTSendDataToBuff(&s1->connectFlag, 1, NUM);
    //保持连接时间
    MQTTSendDataToBuff(&s1->keepAliveTime, 2, NUM);

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

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

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

    /* 等待服务器响应 */
    temp = 0;
    while(mqtt.resultCode == MQTT_RESULT_CODE_INIT)
    {
        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  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, NUM);
    //主题
    MQTTSendDataToBuff((void*)s1->topic, temp, CHAR);
    //报文标识符
    if(s1->QoS)
    {
        MQTTSendDataToBuff(&s1->messageID, 2, NUM);
    }

    /* 有效载荷 */
    temp = strlen(s1->payload);
    MQTTSendDataToBuff((void*)s1->payload, temp, 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)
    {
        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(); 

至此，可以得到

void CONNECT(void)
{
    MQTTConnectStruct_t s1 = {0};
    MQTTConnACKStruct_t s2 = {0};

    s1.connectFlag = 0xC2;//连接标志
    s1.keepAliveTime = 300;//保活时间
    s1.clientID = CLIENT_ID;//客户端标识
    s1.userName = USER_NAME;//用户名
    s1.password = PASSWORD;	//密码
    MQTTConnect(&s1, &s2);//发送连接报文
}

PUBLISH发布消息

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

void PUBLSH(void)
{
    MQTTPublishStruct_t s1;
    MQTTPubACKStruct_t s2;

    s1.DUP = 0;
    s1.QoS = 0;
    s1.RETAIN = 0;
    s1.topic = TOPIC;
    s1.payload = PAYLOAD;
    MQTTPublish(&s1, &s2);
}

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

后续

不足与改进

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

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

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

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

源代码文件下载链接