物联网协议应用

前言

  本文主要介绍一下物联网协议如NTP协议、HTTP协议和MQTT协议的接口使用

一、WIFI简介

  在了解WIFI之前需要了解一下TCP/IP协议和lwIP协议，参考以下链接：https://blog.csdn.net/weixin_44567668/article/details/139619797
  首先lwIP协议是一种专为嵌入式系统设计的轻量级TCP/IP协议栈。lwIP与 TCP/IP体系结构的对应关系：

  回到WIFI，市面上很多都是以太网或者WIFI透传模块，他们有的使用SPI接口，有的使用UART串口来与MCU进行通讯，如ESP8266。但实际上有部分芯片是内嵌WIFI模组的MCU，如ESP32、W601等等。这里以ESP32-S3为例，其内嵌WiFi MAC内核，只需了解它扮演 TCP/IP协议的网络接口层角色即可。如下图所示：

  到这里对WIFI已经有基本的了解，我这里再扩充一下WLAN设备。WLAN 框架是RT-Thread开发的一套用于管理WIFI的中间件。对下连接具体的WIFI驱动，控制 WIFI 的连接断开，扫描等操作。对上承载不同的应用，为应用提供 WIFI 控制，事件，数据导流等操作，为上层应用提供统一的 WIFI 控制接口。WIFI 框架层次图：

• APP：为应用层，是基于WLAN框架的具体应用，如后面的HTTP、MQTT
• WLAN protocol：为协议层，其中上面讲的lwIP协议就处于这一层

二、NTP协议

2.1 NTP简介

  NTP（Network Time Protocol）网络时间协议基于UDP，是用来使计算机时间同步化的一种协议，它可以使计算机对其服务器或时钟源（如石英钟，GPS 等等）做同步化，它可以提供高精准度的时间校正（LAN 上与标准间差小于 1 毫秒，WAN 上几十毫秒），且可介由加密确认的方式来防止恶毒的协议攻击。时间按 NTP 服务器的等级传播。按照离外部 UTC 源的远近把所有服务器归入不同的 Stratum（层）中。
  NTP数据报文格式，如下图所示：

  NTP数据报文格式的各个字段的作用，如下表所示：


  从上表可知，NTP 报文的字段非常多，这些字段并不是每一个都必须设置的，可以根据项目的需要来构建 NTP 请求报文。

2.2 NTP实现

  由上可以知道获取 NTP 实时时间步骤了：

① 以 UDP 协议连接阿里云 NTP 服务器
② 发送 NTP 报文到阿里云 NTP 服务器
③ 获取阿里云 NTP服务器返回的数据，取第 40 位到 43 位的十六进制数值。
④ 把 40 位到 43 位的十六进制数值转成十进制
⑤ 把十进制数值减去1900-1970 的时间差（2208988800 秒）
⑥ 数值转成年月日时分秒

1. lwip_demo.h头文件
     主要创建两个结构体，一个用来获取参数，一个用来显示时间

#define NTP_DEMO_RX_BUFSIZE   2000  /* 定义udp最大接收数据长度 */
#define NTP_DEMO_PORT         123   /* 定义udp连接的本地端口号 */typedef struct _NPTformat
{char    version;            /* 版本号 */char    leap;               /* 时钟同步 */char    mode;               /* 模式 */char    stratum;            /* 系统时钟的层数 */char    poll;               /* 更新间隔 */signed char  precision;     /* 精密度 */unsigned int   rootdelay;   /* 本地到主参考时钟源的往返时间 */unsigned int   rootdisp;    /* 统时钟相对于主参考时钟的最大误差 */char    refid;              /* 参考识别码 */unsigned long long  reftime;/* 参考时间 */unsigned long long  org;    /* 开始的时间戳 */unsigned long long  rec;    /* 收到的时间戳 */unsigned long long  xmt;    /* 传输时间戳 */
} NPTformat;typedef struct _DateTime  /*此结构体定义了NTP时间同步的相关变量*/
{int  year;        /* 年 */int  month;       /* 月 */int  day;         /* 天 */int  hour;        /* 时 */int  minute;      /* 分 */int  second;      /* 秒 */
} DateTime;#define SECS_PERDAY     86400UL         /* 一天中的几秒钟 = 60*60*24 */
#define UTC_ADJ_HRS     8               /* SEOUL : GMT+8（东八区北京）  */
#define EPOCH           1900            /* NTP 起始年  */
#define HOST_NAME  "ntp1.aliyun.com"    /*阿里云NTP服务器域名 */

2. lwip_demo.c源文件

#define NTP_TIMESTAMP_DELTA  2208988800UL
const char g_days[12] = {31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
NPTformat g_ntpformat;                                                        /* NT数据包结构体 */
DateTime g_nowdate;                                                           /* 时间结构体 */
uint8_t g_ntp_message[48];                                                    /* 发送数据包的缓存区 */  
uint8_t g_ntp_demo_recvbuf[NTP_DEMO_RX_BUFSIZE];                              /* NTP接收数据缓冲区 */uint8_t g_lwip_time_buf[100];/***@brief     计算日期时间*@param     secondsUTC 世界标准时间*@retval    无
*/
void lwip_calc_date_time(unsigned long long time)
{unsigned int Pass4year;int hours_per_year;if (time <= 0){time = 0;}g_nowdate.second = (int)(time % 60);   /* 取秒时间 */time /= 60;g_nowdate.minute = (int)(time % 60);   /* 取分钟时间 */time /= 60;g_nowdate.hour = (int)(time % 24);     /* 小时数 */Pass4year = ((unsigned int)time / (1461L * 24L));/* 取过去多少个四年，每四年有 1461*24 小时 */g_nowdate.year = (Pass4year << 2) + 1970;    /* 计算年份 */time %= 1461 * 24;     /* 四年中剩下的小时数 */for (;;)               /* 校正闰年影响的年份，计算一年中剩下的小时数 */{hours_per_year = 365 * 24;         /* 一年的小时数 */if ((g_nowdate.year & 3) == 0) /* 判断闰年 */{hours_per_year += 24;          /* 是闰年，一年则多24小时，即一天 */}if (time < hours_per_year){break;}g_nowdate.year++;time -= hours_per_year;}time /= 24;   /* 一年中剩下的天数 */time++;       /* 假定为闰年 */if ((g_nowdate.year & 3) == 0)      /* 校正闰年的误差，计算月份，日期 */{if (time > 60){time--;}else{if (time == 60){g_nowdate.month = 1;g_nowdate.day = 29;return ;}}}for (g_nowdate.month = 0; g_days[g_nowdate.month] < time; g_nowdate.month++)   /* 计算月日 */{time -= g_days[g_nowdate.month];}g_nowdate.day = (int)(time);return;}/***@brief     从NTP服务器获取时间*@param     buf：存放缓存*@param     idx：定义存放数据起始位置*@retval    无
*/
void lwip_get_seconds_from_ntp_server(uint8_t *buf, uint16_t idx)
{unsigned long long atk_seconds = 0; uint8_t i = 0;for (i = 0; i < 4; i++)  /* 获取40~43位的数据 */{atk_seconds = (atk_seconds << 8) | buf[idx + i]; /* 把40~43位转成16进制再转成十进制 */}atk_seconds -= NTP_TIMESTAMP_DELTA;/* 减去减去1900-1970的时间差（2208988800秒） */lwip_calc_date_time(atk_seconds);       /* 由UTC时间计算日期 */
}/***@brief     初始化NTP Client信息*@param     无*@retval    无
*/
void lwip_ntp_client_init(void)
{uint8_t flag;g_ntpformat.leap = 0;           /* leap indicator */g_ntpformat.version = 3;        /* version number */g_ntpformat.mode = 3;           /* mode */g_ntpformat.stratum = 0;        /* stratum */g_ntpformat.poll = 0;           /* poll interval */g_ntpformat.precision = 0;      /* precision */g_ntpformat.rootdelay = 0;      /* root delay */g_ntpformat.rootdisp = 0;       /* root dispersion */g_ntpformat.refid = 0;          /* reference ID */g_ntpformat.reftime = 0;        /* reference time */g_ntpformat.org = 0;            /* origin timestamp */g_ntpformat.rec = 0;            /* receive timestamp */g_ntpformat.xmt = 0;            /* transmit timestamp */flag = (g_ntpformat.version << 3) + g_ntpformat.mode; /* one byte Flag */memcpy(g_ntp_message, (void const *)(&flag), 1);
}/*** @brief  lwip_demo程序入口* @param  无* @retval 无
*/
void lwip_demo(void)
{err_t err;static struct netconn *udpconn;static struct netbuf  *recvbuf;static struct netbuf  *sentbuf;ip_addr_t destipaddr;uint32_t data_len = 0;struct pbuf *q;lwip_ntp_client_init();/* 第一步：创建udp控制块 */udpconn = netconn_new(NETCONN_UDP);/* 定义接收超时时间 */udpconn->recv_timeout = 10;if (udpconn != NULL) /* 判断创建控制块释放成功 */{/* 第二步：绑定控制块、本地IP和端口 */err = netconn_bind(udpconn, IP_ADDR_ANY, NTP_DEMO_PORT);/* 域名解析 */netconn_gethostbyname((char *)(HOST_NAME), &(destipaddr));/* 第三步：连接或者建立对话框 */netconn_connect(udpconn, &destipaddr, NTP_DEMO_PORT); /* 连接到远端主机 */if (err == ERR_OK) /* 绑定完成 */{while (1){sentbuf = netbuf_new();netbuf_alloc(sentbuf, 48);memcpy(sentbuf->p->payload, (void *)g_ntp_message, sizeof(g_ntp_message));err = netconn_send(udpconn, sentbuf); /* 将sentbuf中的数据发送出去 */if (err != ERR_OK){printf("发送失败\r\n");netbuf_delete(sentbuf); /* 删除buf */}netbuf_delete(sentbuf);      /* 删除buf *//* 第五步：接收数据 */netconn_recv(udpconn, &recvbuf);vTaskDelay(1000);     /* 延时1s */if (recvbuf != NULL)  /* 接收到数据 */{memset(g_ntp_demo_recvbuf, 0, NTP_DEMO_RX_BUFSIZE); /*数据接收缓冲区清零 */for (q = recvbuf->p; q != NULL; q = q->next) /*遍历完整个pbuf链表 */{/* 判断要拷贝到UDP_DEMO_RX_BUFSIZE中的数据是否大于UDP_DEMO_RX_BUFSIZE的剩余空间，如果大于 *//* 的话就只拷贝UDP_DEMO_RX_BUFSIZE中剩余长度的数据，否则的话就拷贝所有的数据 */if (q->len > (NTP_DEMO_RX_BUFSIZE - data_len)) memcpy(g_ntp_demo_recvbuf + data_len, q->payload, (NTP_DEMO_RX_BUFSIZE - data_len)); /* 拷贝数据 */else memcpy(g_ntp_demo_recvbuf + data_len, q->payload, q->len);data_len += q->len;if (data_len > NTP_DEMO_RX_BUFSIZE) break;          /* 超出TCP客户端接收数组,跳出 */}data_len = 0;                                           /* 复制完成后data_len要清零 */lwip_get_seconds_from_ntp_server(g_ntp_demo_recvbuf,40);   /* 从NTP服务器获取时间 */printf("北京时间：%02d-%02d-%02d %02d:%02d:%02d\r\n",  g_nowdate.year, g_nowdate.month + 1,g_nowdate.day,g_nowdate.hour + 8,g_nowdate.minute,g_nowdate.second);sprintf((char*)g_lwip_time_buf,"BJ time:%02d-%02d-%02d %02d:%02d:%02d", g_nowdate.year, g_nowdate.month + 1,g_nowdate.day,g_nowdate.hour + 8,g_nowdate.minute,g_nowdate.second);lcd_show_string(5, 170, lcddev.width, 16, 16, (char*)g_lwip_time_buf, RED);netbuf_delete(recvbuf);             /* 删除buf */}else vTaskDelay(5);                     /* 延时5ms */}}else printf("NTP绑定失败\r\n");}else printf("NTP连接创建失败\r\n");
}

  在此文件下定义了四个函数，这些函数的作用如下表所示：

三、HTTP协议

3.1 HTTP协议简介

  HTTP（Hypertext Transfer Protocol）协议，即超文本传输协议，是用于从万维网（WWW:World Wide Web ）服务器传输超文本到本地浏览器的传送协议。HTTP 协议是基于TCP/IP 协议的网络应用层协议。默认端口为80端口。HTTP 协议是一种请求/响应式的协议。一个客户端与服务器建立连接之后，发送一个请求给服务器。服务器接收到请求之后，通过接收到的信息判断响应方式，并且给予客户端相应的响应，完成整个 HTTP数据交互流程。

  HTTP定义了与服务器交互的不同方法，其最基本的方法是 GET、PORT 和 HEAD。如下图所示。

• GET：从服务端获取数据。
• PORT：向服务器传送数据。
• HEAD：检测一个对象是否存在。

  互联网通过URL来定位，URL全称是 Uniform Resource Locator，是互联网上用来标识某一处资源的绝对地址，大部分 URL 都会遵循 URL 的语法，一个 URL 的组成有多个不同的组件，一个 URL的通用格式如下：

<scheme>://<user>:<password>@<host>:<port>/<path>;<params>?<query>#<frag>

  HTTP 报文是由 3 个部分组成，分别是：对报文进行描述的“起始行”，包含属性的“首部”，以及可选的“数据主体”，对于请求报文与应答报文，只有“起始行”的格式是不一样的。起始行和首部就是由行分隔的 ASCII 文本组成，每行都以由两个字符组成的行终止序列作为结束，其中包括一个回车符（ASCII 码 13）和一个换行符（ASCII 码 10）， 这个行终止序列可以写做 CRLF。

# HTTP请求报文
<method> <request-URL> <version> //起始行
<headers> 						//首部<entity-body> 					//数据主体# HTTP应答报文
<version> <status> <reason-phrase> //起始行
<headers> 							//首部<entity-body> 						//数据主体

  下面就对这两种 HTTP 报文的各个部分简单描述一下：

• 方法（method）：HTTP 请求报文的起始行以方法作为开始，方法用来告知服务器要做些什么，常见的方法有 GET、POST、HEAD 等，比如“GET /forum.php HTTP/1.1” 使用的就是 GET 方法。
• 请求 URL（request-URL）：指定了所请求的资源。
• 版本（version）：指定报文所使用的 HTTP 协议版本，其中指定了主要版本号， 指定了次要版本号，它们都是整数，其格式如下：

HTTP/<major>.<minor>

• 状态码（status）：这是在 HTTP 应答报文中使用的，状态码是在每条响应报文的起始行中返回的一个数字码，描述了请求过程中所发送的情况，比如成功、失败等，不同的状态码有不同的含义，具体见表格
  
• 原因短语（reason-phrase）：这其实是给我们看的原因短语，因为数字是不够直观，它只是状态码的一个文本形式表达而已。
• 首部（header）：HTTP 报文可以有 0 个、1 个或者多个首部，HTTP 首部字段向请求和响应报文中添加了一些附加信息，从本质上来说，它们是一个<名字：值>对，每个首部都包含一个名字，紧跟着一个冒号“:”，然后是一个可选的空格，接着是一个值，最后以 CRLF 结束，比如“Host: www.firebbs.cn”就是一个首部。
• 数据主体（entity-body）：这部分包含一个由任意数据组成的数据块，其实这与我们前面所讲的报文数据区域是一样的，用于携带数据，HTTP 报文可以承载很多类型的数字数据：图片、视频、音频、HTML 文档、软件应用程 序等。

3.2 HTTP服务器

  HTTP协议可以应用在客户端，也可以在服务器端，在客户端可以用来获取服务器数据，比如从服务器下载固件进行升级。也可以用在服务器端，那样我们可以做一个简单网页来访问控制单片机。同上一个例程一样，新建一个任务调用函数lwip_demo()

/* HTTP报头总是以响应码开头(例如HTTP/1.1 200 OK)和一个内容类型，以便客户端知道接下来是什么，然后是一个空行: */
/* 浏览器响应数据类型为文本数据 */
static const char http_html_hdr[] = "HTTP/1.1 200 OK\r\nContent-type: text/html\r\n\r\n";
static const char http_index_html[] = "<!DOCTYPE html>"\"<html>"\"<head>"\"<title> Webserver实验 </title>"\"<meta http-equiv='Content-Type' content='text/html; charset=GB2312'/>"\"</head>"\"<body>"\"<h1>http server</h1>"\"<div class='label' >"\"<label>LED State:</label>"\"</div>"\"<div class='checkboxes'>"\"<input type='checkbox' name='led1' value='1' />打开&nbsp;&nbsp;&nbsp;<input type='checkbox' name='led1' value='2' />关闭"\"</div>"\"<br>"\"<br>"\"<div class='label'>"\"<label>BEEP State:</label>"\"</div>"\"<div class='checkboxes'>"\"<input type='checkbox' name='led1' value='1' />打开&nbsp;&nbsp;&nbsp;<input type='checkbox' name='led1' value='2' />关闭"\"</div>"\"<br>"\"<br>"\"<input type='submit' class='sendbtn' value='发送'>"\"<br>"\"</body>"\"</html>";/*** @brief       寻找指定字符位置* @param       buf   缓冲区指针* @param       name  寻找字符* @retval      返回字符的地址*/
char *lwip_data_locate(char *buf, char *name)
{char *p;p = strstr((char *)buf, name);if (p == NULL){return NULL;}p += strlen(name);return p;
}/*** @brief       服务HTTP线程中接受的一个HTTP连接* @param       conn   netconn控制块* @retval      无*/
static void lwip_server_netconn_serve(struct netconn *conn)
{struct netbuf *inbuf;char *buf;u16_t buflen;err_t err;char *ptemp;/* 从端口读取数据，如果那里还没有数据，则阻塞。我们假设请求(我们关心的部分)在一个netbuf中 */err = netconn_recv(conn, &inbuf);if (err == ERR_OK){netbuf_data(inbuf, (void **)&buf, &buflen);/* 这是一个HTTP GET命令吗?只检查前5个字符，因为GET还有其他格式，我们保持简单)*/if (buflen >= 5 &&buf[0] == 'G' &&buf[1] == 'E' &&buf[2] == 'T' &&buf[3] == ' ' &&buf[4] == '/' ){start_html:/* 发送HTML标题从大小中减去1，因为我们没有在字符串中发送\0NETCONN_NOCOPY:我们的数据是常量静态的，所以不需要复制它 */netconn_write(conn, http_html_hdr, sizeof(http_html_hdr) - 1, NETCONN_NOCOPY);/* 发送我们的HTML页面 */netconn_write(conn, http_index_html, sizeof(http_index_html) - 1, NETCONN_NOCOPY);}else if(buflen>=8&&buf[0]=='P'&&buf[1]=='O'&&buf[2]=='S'&&buf[3]=='T'){ptemp = lwip_data_locate((char *)buf, "led1=");if (ptemp != NULL){if (*ptemp == '1')    /* 查看led1的值。为1则灯亮，为2则灭，此值与HTML网页中设置有关 */{LED0(0);  /* 点亮LED1 */}else{LED0(1);    /* 熄灭LED1 */}}ptemp = lwip_data_locate((char *)buf, "beep=");    /* 查看beep的值。为3则灯亮，为4则灭，此值与HTML网页中设置有关 */if (ptemp != NULL ){if (*ptemp == '3'){/* 打开蜂鸣器 */}else{/* 关闭蜂鸣器 */}}goto start_html;}}/* 关闭连接(服务器在HTTP中关闭) */netconn_close(conn);/* 删除缓冲区(netconn_recv给我们所有权，所以我们必须确保释放缓冲区) */netbuf_delete(inbuf);
}/**
* @brief  lwip_demo程序入口
* @param  无
* @retval 无
*/
void lwip_demo(void)
{struct netconn *conn, *newconn;err_t err;/* 创建一个新的TCP连接句柄 *//* 使用默认IP地址绑定到端口80 (HTTP) */conn = netconn_new(NETCONN_TCP);netconn_bind(conn, IP_ADDR_ANY, 80);/* 将连接置于侦听状态 */netconn_listen(conn);do{err = netconn_accept(conn, &newconn);if (err == ERR_OK){lwip_server_netconn_serve(newconn);//调用HTTP服务器子程序netconn_delete(newconn);}}while (err == ERR_OK);netconn_close(conn);netconn_delete(conn);
}

• lwip_demo()：建立 TCP 连接
• lwip_data_locate()：寻找指定字符位置
• lwip_server_netconn_serve()：服务 HTTP 线程中接受的一个HTTP连接，主要分为三步：
  • 当浏览器输入IP地址并且回车确认时，程序调用函数netconn_write把网页数据发送到浏览器当中
  • 当网页发送一个PORT命令时，程序调用函数lwip_data_locate判断触发源，判断完成之后根据触发源来执行相应的动作
  • 程序执行goto语句重新发送网页字符串到网页当中，这个步骤相当于更新网页，网页样式如下：

  其中网页格式为HTML，具体语法可参考：

• HTML5超文本标记语言：https://blog.csdn.net/weixin_44567668/article/details/125626370
• CSS3层叠样式表：https://blog.csdn.net/weixin_44567668/article/details/132521477

四、MQTT协议

4.1 MQTT协议简介

  MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一个基于客户端-服务器的消息发布/订阅（publish/subscribe）传输协议，该协议构建于 TCP/IP 协议上，由 IBM 在 1999 年发布。

4.1.1 MQTT通信模型

  实现 MQTT 协议需要：客户端和服务器端 MQTT 协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者，如下图所示

  MQTT 传输的消息分为：主题（Topic）和消息的内容（payload）两部分

• Topic：可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）。
• Payload：可以理解为消息的内容，是指订阅者具体要使用的内容。

  正常MCU的MQTT应用于客户端，用来向服务器发送或者接收数据

4.1.2 MQTT协议实现原理

  要在客户端与代理服务端建立一个 TCP 连接，建立连接的过程是由客户端主动发起的，代理服务一直是处于指定端口的监听状态，当监听到有客户端要接入的时候，就会立刻去处理。客户端在发起连接请求时，携带客户端 ID、账号、密码、心跳间隔时间等数据。代理服务收到后检查自己的连接权限配置中是否允许该账号密码连接，如果允许则建立会话标识并保存，绑定客户端 ID 与会话，并记录心跳间隔时间（判断是否掉线和启动遗嘱时用）和遗嘱消息等，然后回发连接成功确认消息给客户端，客户端收到连接成功的确认消息后，进入下一步（通常是开始订阅主题，如果不需要订阅则跳过）。如下图所示：

  客户端将需要订阅的主题经过 SUBSCRIBE 报文发送给代理服务，代理服务则将这个主题记录到该客户端 ID 下（以后有这个主题发布就会发送给该客户端），然后回复确认消息SUBACK 报文，客户端接到 SUBACK 报文后知道已经订阅成功，则处于等待监听代理服务推送的消息，也可以继续订阅其他主题或发布主题，如下图所示：

  当某一客户端发布一个主题到代理服务后，代理服务先回复该客户端收到主题的确认消息，该客户端收到确认后就可以继续自己的逻辑了。但这时主题消息还没有发给订阅了这个主题的客户端，代理要根据质量级别（QoS）来决定怎样处理这个主题。所以这里充分体现了是MQTT 协议是异步通信模式，不是立即端到端反应的，如下图所示：

• 如果发布和订阅时的质量级别 QoS 都是至多一次，那代理服务则检查当前订阅这个主题的客户端是否在线，在线则转发一次，收到与否不再做任何处理。这种质量对系统压力最小。
• 如果发布和订阅时的质量级别 QoS 都是至少一次，那要保证代理服务和订阅的客户端都有成功收到才可以，否则会尝试补充发送（具体机制后面讨论）。这也可能会出现同一主题多次重复发送的情况。这种质量对系统压力较大。
• 如果发布和订阅时的质量级别 QoS 都是只有一次，那要保证代理服务和订阅的客户端都有成功收到，并只收到一次不会重复发送（具体机制后面讨论）。这种质量对系统压力最大。

4.1.3 MQTT 控制报文

1. 固定报头
     MQTT 协议工作在 TCP 协议之上，因为客户端和服务器都是应用层，那么必然需要一种协议在两者之间进行通信，那么随之而来的就是 MQTT 控制报文， MQTT 控制报文有3个部分组成，分别是固定报头（fixed header）、可变报头（variable header）、有效荷载（数据区域 payload）。固定报头，所有的 MQTT 控制报文都包含，可变报头与有效载荷是部分 MQTT 控制报文包含。固定报头占据两字节的空间，具体见图
   
     固定报头的第一个字节分为控制报文的类型（4bit），以及控制报文类型的标志位，控制类型共有 14 种，其中0与15被系统保留出来，其他的类型具体见表格
   
     固定报头的 bit0-bit3 为标志位，依照报文类型有不同的含义，事实上，除了 PUBLISH类型报文以外，其他报文的标志位均为系统保留，PUBLISH 报文的第一字节 bit3 是控制报文的重复分发标志（DUP），bit1-bit2 是服务质量等级，bit0 是 PUBLISH 报文的保留标志，用于标识 PUBLISH 是否保留，当客户端发送一个 PUBLISH 消息到服务器，如果保留标识位置 1，那么服务器应该保留这条消息，当一个新的订阅者订阅这个主题的时候，最后保留的主题消息应被发送到新订阅的用户。
     固定报头的第二个字节开始是剩余长度字段，是用于记录剩余报文长度的，表示当前的消息剩余的字节数，包括可变报头和有效载荷区域（如果存在），但剩余长度不包括用于编码剩余长度字段本身的字节数。
     剩余长度字段使用一个变长度编码方案，对小于 128 的值它使用单字节编码，而对于更大的数值则按下面的方式处理：每个字节的低 7 位用于编码数据长度，最高位（bit7）用于标识剩余长度字段是否有更多的字节，且按照大端模式进行编码，因此每个字节可以编码 128 个数值和一个延续位，剩余长度字段最大可拥有 4 个字节。

2. 可变报头
     可变报头并不是所有的 MQTT 报文都带有的（比如 PINGREQ 心跳请求与 PINGRESP心跳响应报文就没有可变报头），只有某些报文才拥有可变报头，它在固定报头和有效负载之间，可变报头的内容会根据报文类型的不同而有所不同，但可变报头的报文标识符（Packet Identifier）字段存在于在多个类型的报文里，而有一些报文又没有报文标识符字段，具体见表格
   
     报文标识符结构具体见图
   
     因为对于不同的报文，可变报头是不一样的，下面就简单讲解几个报文的可变报头

• CONNECT
    在一个会话中，客户端只能发送一次 CONNECT 报文，它是客户端用于请求连接服务器的报文，常称之为连接报文，如果客户端发送多次连接报文，那么服务端必须将客户端发送的第二个 CONNECT 报文当作协议违规处理并断开客户端的连接。CONNECT 报文的可变报头包含四个字段：协议名（Protocol Name）、协议级别（Protocol Level）、连接标志（Connect Flags）以及保持连接（Keep Alive）字段。协议名是 MQTT 的 UTF-8 编码的字符串，其中还包含用于记录协议名长度的两字节字段 MSB 与 LSB。
    在协议名之后的是协议级别，MQTT 协议使用 8 位的无符号值表示协议的修订版本，对于 MQTT3.1 版的协议，协议级别字段的值是 3(0x03)，而对于 MQTT3.1.1 版的协议，协议级别字段的值是 4(0x04)。如果服务器发现连接报文中的协议级别字段是不支持的协议级别，服务端必须给发送一个返回码为 0x01（不支持的协议级别）的 CONNACK 响应连接报文，然后终止客户端的连接请求。连接标志字段涉及的内容比较多，它在协议级别之后使用一个字节表示，但分成很多个标志位，具体见图
  
  • bit0：是 MQTT 保留的标志位，在连接过程中，服务器会检测连接标志的 bit0 是否为 0，如果不为 0 则服务器任务这个连接报文是不合法的，会终止连接请求。
  • bit1：是清除会话标志 Clean Session，如果清除会话标志设置为 1，那么客户端不会收到旧的应用消息，清除会话标志设置为 0 的客户端在重新连接后会收到所有在它连接断开期间（其他发布者）发布的 QoS1 和 QoS2 级别的消息。因此，要确保不丢失连接断开期间的消息，需要使用 QoS1 或 QoS2 级别，同时将清除会话标志设置为 0。
  • bit2：是遗嘱标志 Will Flag，如果该位被设置为 1，表示如果客户端与服务器建立了会话，遗嘱消息（Will Message）将必须被存储在服务器中，当这个客户端断开连接的时候，遗嘱消息将被发送到订阅这个会话主题的所有订阅者，这个消息是很有用的，我们可以知道这个设备的状况，它是否已经掉线了，以备启动备用方案，当然，想要不发送遗嘱消息也是可以的，只需要让服务器端收到 DISCONNECT 报文时删除这个遗嘱消息即可。
  • bit3-bit4：用于指定发布遗嘱消息时使用的服务质量等级，与其他消息的服务质量是一样的，遗嘱 QoS 的值可以等于 0(0x00)，1(0x01)，2(0x02)，当然，使用遗嘱消息的前提是遗嘱标志位为 1。
  • bit5：表示遗嘱保留标志位，当客户端意外断开连接时，如果 Will Retain 置一，那么服务器必须将遗嘱消息当作保留消息发布，反之则无需保留。
  • bit6：是密码标志位 Password Flag，如果密码标志被设置为 0，有效载荷中不能包含密码字段，反之则必须包含密码字段。
  • bit7：是用户名标志位 User Name Flag，如果用户名标志被设置为 0，有效载荷中不能包
    含用户名字段，反之则必须包含用户名字段。

  总的来说，整个 CONNECT 报文可变报头的内容如下：

• CONNACK
    它是由连接确认标志字段（Connect Acknowledge Flags）与连接返回码字段 （Connect Return code）组成，各占用 1 个字节。它的第 1 个字节是 连接确认标志字段，bit1-bit7 是保留位且必须设置为 0， bit0 是当前会话（Session Present）标志位。它的第 2 个字节是返回码字段，如果服务器收到一个 CONNECT 报文，但出于某些原因无法处理它，服务器会返回一个包含返回码的 CONNACK 报文。如果服务器返回了一个返回码字段是非 0 的 CONNACK 报文，那么它必须关闭网络连接，返回码描述具体见表格
  
    如果服务端收到清理会话（CleanSession）标志为 1 的连接，除了将 CONNACK报文中的返回码设置为 0 之外，还必须将 CONNACK 报文中的当前会话设置（Session Present）标志为 0。那么总的来说，CONNACK 报文的可变报头部分内容具体见图
  

4.2 移植MQTT协议

4.2.1 方法一

1. MQTT协议移植
     由第一章可知MQTT是应用层协议，基于TCP协议，所以我们就使用 Socket API 来进行移植。首先下载 MQTT 的库：https://github.com/eclipse/paho.mqtt.embedded-c
     然后在工程文件新建一个MQTT子文件夹，将MQTT库文件的MQTTPacket\src目录下的文件添加到工程目录MQTT文件夹，再将MQTTPacket\samples目录下的transport.c、transport.h添加到这个文件夹下
   
     我们把这些文件加入我们的工程之中，并且指定头文件路径，然后实现transport.c文件的移植层接口

#include "transport.h"
#include "lwip/opt.h"
#include "lwip/arch.h"
#include "lwip/api.h"
#include "lwip/inet.h"
#include "lwip/sockets.h"
#include "string.h"static int mysock;/************************************************************************
** 函数名称: transport_sendPacketBuffer									
** 函数功能: 以TCP方式发送数据
** 入口参数: unsigned char* buf：数据缓冲区
**           int buflen：数据长度
** 出口参数: <0发送数据失败							
************************************************************************/
int32_t transport_sendPacketBuffer( uint8_t* buf, int32_t buflen)
{int32_t rc;rc = write(mysock, buf, buflen);return rc;
}/************************************************************************
** 函数名称: transport_getdata									
** 函数功能: 以阻塞的方式接收TCP数据
** 入口参数: unsigned char* buf：数据缓冲区
**           int count：数据长度
** 出口参数: <=0接收数据失败									
************************************************************************/
int32_t transport_getdata(uint8_t* buf, int32_t count)
{int32_t rc;//这个函数在这里不阻塞rc = recv(mysock, buf, count, 0);return rc;
}/************************************************************************
** 函数名称: transport_open									
** 函数功能: 打开一个接口，并且和服务器 建立连接
** 入口参数: char* servip:服务器域名
**           int   port:端口号
** 出口参数: <0打开连接失败										
************************************************************************/
int32_t transport_open(int8_t* servip, int32_t port)
{int32_t *sock = &mysock;int32_t ret;
//	int32_t opt;struct sockaddr_in addr;//初始换服务器信息memset(&addr,0,sizeof(addr));addr.sin_len = sizeof(addr);addr.sin_family = AF_INET;//填写服务器端口号addr.sin_port = PP_HTONS(port);//填写服务器IP地址addr.sin_addr.s_addr = inet_addr((const char*)servip);//创建SOCK*sock = socket(AF_INET,SOCK_STREAM,0);//连接服务器 ret = connect(*sock,(struct sockaddr*)&addr,sizeof(addr));if(ret != 0){//关闭链接close(*sock);//连接失败return -1;}//连接成功,设置超时时间1000ms
//	opt = 1000;
//	setsockopt(*sock,SOL_SOCKET,SO_RCVTIMEO,&opt,sizeof(int));//返回套接字return *sock;
}/************************************************************************
** 函数名称: transport_close									
** 函数功能: 关闭套接字
** 入口参数: unsigned char* buf：数据缓冲区
**           int buflen：数据长度
** 出口参数: <0发送数据失败							
************************************************************************/
int transport_close(void)
{int rc;
//	rc = close(mysock);rc = shutdown(mysock, SHUT_WR);rc = recv(mysock, NULL, (size_t)0, 0);rc = close(mysock);return rc;
}

• 首先添加相关头文件
• transport_sendPacketBuffer()函数：是 MQTT发送数据函数，这个函数必须以TCP协议发送数据，参数 buf指定数据缓冲区，buflen指定了数据长度，调用write()函数进行发送数据，并且返回发送状态。
• transport_getdata()函数：是 MQTT 接收数据的函数，需要我们用Socket API 获取接收到的数据，参数 buf 指定数据缓冲区，count 指定了获取数据长度，我们只要调用 recv()将数据获取回来即可。
• transport_getdata()函数：是 MQTT 接收数据的函数，需要我们用Socket API 获取接收到的数据，参数 buf 指定数据缓冲区，count 指定了获取数据长度，我们只要调用 recv()将数据获取回来即可。
• transport_close()：是 MQTT 与服务器断开的时候会调用的函数，它用来关闭一个套接字的。

2. cJSON移植
   json是一种完全独立于编程语言的文本格式，用来存储和表示数据，具有以下特点：

• 简洁和层次结构清晰
• 易于人阅读和编写
• 易于机器解析和生成，并有效地提升网络传输效率

  cJSON是一个用于解析JSON包的C语言库，库文件为cJSON.c和cJSON.h， 所有的实现都在这两个文件中。cJSON的移植很简单，首先我们首先下载到 cJSON 的源码文件：https://github.com/DaveGamble/cJSON
  然后在文件目录下找到cJSON.c和cJSON.h，将它们拷贝到我们的工程目录下的cJSON 文件夹下（如果没有就创建它），然后添加到工程中，并且指定头文件路径即可。由于LWIP是包含RTOS的，所以还需要将相关头文件添加进来，并且cJSON 中的动态内存分配、释放函数就需要配合操作系统的动态内存分配函数与释放函数

//#include "FreeRTOS.h"		//如果是FreeRTOS添加#if defined(_MSC_VER)
/* work around MSVC error C2322: '...' address of dillimport '...' is not static */
static void * CJSON_CDECL internal_malloc(size_t size)
{
//    return malloc(size);return pvPortMalloc(size);
}
static void CJSON_CDECL internal_free(void *pointer)
{
//    free(pointer);vPortFree(pointer);
}
static void * CJSON_CDECL internal_realloc(void *pointer, size_t size)
{
//    return realloc(pointer, size);return NULL;
}
#else
#define internal_malloc pvPortMalloc
#define internal_free vPortFree
#define internal_realloc 
#endif

  然后再对cJSON进行了封装，包含cJSON格式数据的初始化、更新、解析等，当然大家也可以自行封装使用，我们创建一个cJSON_Process.c文件，并添加以下代码

#include "cJSON_Process.h"
#include "main.h"/********************************************************************                          变量声明                               *******************************************************************/cJSON* cJSON_Data_Init(void)
{cJSON* cJSON_Root = NULL;    //json根节点cJSON_Root = cJSON_CreateObject();   /*创建项目*/if(NULL == cJSON_Root){return NULL;}cJSON_AddStringToObject(cJSON_Root, NAME, DEFAULT_NAME);  /*添加元素  键值对*/cJSON_AddNumberToObject(cJSON_Root, TEMP_NUM, DEFAULT_TEMP_NUM);cJSON_AddNumberToObject(cJSON_Root, HUM_NUM, DEFAULT_HUM_NUM);char* p = cJSON_Print(cJSON_Root);  /*p 指向的字符串是json格式的*///  PRINT_DEBUG("%s\n",p);vPortFree(p);p = NULL;return cJSON_Root;}
uint8_t cJSON_Update(const cJSON * const object,const char * const string,void *d)
{cJSON* node = NULL;    //json根节点node = cJSON_GetObjectItem(object,string);if(node == NULL)return NULL;if(cJSON_IsBool(node)){int *b = (int*)d;
//    printf ("d = %d",*b);cJSON_GetObjectItem(object,string)->type = *b ? cJSON_True : cJSON_False;
//    char* p = cJSON_Print(object);    /*p 指向的字符串是json格式的*/return 1;}else if(cJSON_IsString(node)){cJSON_GetObjectItem(object,string)->valuestring = (char*)d;
//    char* p = cJSON_Print(object);    /*p 指向的字符串是json格式的*/return 1;}else if(cJSON_IsNumber(node)){double *num = (double*)d;
//    printf ("num = %f",*num);
//    cJSON_GetObjectItem(object,string)->valueint = (double)*num;cJSON_GetObjectItem(object,string)->valuedouble = (double)*num;
//    char* p = cJSON_Print(object);    /*p 指向的字符串是json格式的*/return 1;}elsereturn 1;
}void Proscess(void* data)
{PRINT_DEBUG("开始解析JSON数据");cJSON *root,*json_name,*json_temp_num,*json_hum_num;root = cJSON_Parse((char*)data); //解析成json形式json_name = cJSON_GetObjectItem( root , NAME);  //获取键值内容json_temp_num = cJSON_GetObjectItem( root , TEMP_NUM );json_hum_num = cJSON_GetObjectItem( root , HUM_NUM );PRINT_DEBUG("name:%s\n temp_num:%f\n hum_num:%f\n",json_name->valuestring,json_temp_num->valuedouble,json_hum_num->valuedouble);cJSON_Delete(root);  //释放内存 
}

3. MQTT客户端实现
     然后我们在工程中实现两个线程，一个是 MQTT 发送线程，另一个是 MQTT 接收线程。创建一个mqttclient.c文件，然后加入代码

#include "mqttclient.h"
#include "transport.h"
#include "MQTTPacket.h"
#include "FreeRTOS.h"
#include "task.h"
#include "string.h"
#include "sockets.h"#include "lwip/opt.h"#include "lwip/sys.h"
#include "lwip/api.h"#include "lwip/sockets.h"#include "cJSON_Process.h"
#include "bsp_dht11.h"/******************************* 全局变量声明 ************************************/
/** 当我们在写应用程序的时候，可能需要用到一些全局变量。*/
extern QueueHandle_t MQTT_Data_Queue;//定义用户消息结构体
MQTT_USER_MSG  mqtt_user_msg;int32_t MQTT_Socket = 0;void deliverMessage(MQTTString *TopicName,MQTTMessage *msg,MQTT_USER_MSG *mqtt_user_msg);/************************************************************************
** 函数名称: MQTT_Connect								
** 函数功能: 初始化客户端并登录服务器
** 入口参数: int32_t sock:网络描述符
** 出口参数: >=0:发送成功 <0:发送失败
** 备    注: 
************************************************************************/
uint8_t MQTT_Connect(void)
{MQTTPacket_connectData data = MQTTPacket_connectData_initializer;uint8_t buf[200];int buflen = sizeof(buf);int len = 0;data.clientID.cstring = CLIENT_ID;                   //随机data.keepAliveInterval = KEEPLIVE_TIME;         //保持活跃data.username.cstring = USER_NAME;              //用户名data.password.cstring = PASSWORD;               //秘钥data.MQTTVersion = MQTT_VERSION;                //3表示3.1版本，4表示3.11版本data.cleansession = 1;//组装消息len = MQTTSerialize_connect((unsigned char *)buf, buflen, &data);//发送消息transport_sendPacketBuffer(buf, len);/* 等待连接响应 */if (MQTTPacket_read(buf, buflen, transport_getdata) == CONNACK){unsigned char sessionPresent, connack_rc;if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0){PRINT_DEBUG("无法连接，错误代码是: %d！\n", connack_rc);return Connect_NOK;}else {PRINT_DEBUG("用户名与秘钥验证成功，MQTT连接成功！\n");return Connect_OK;}}elsePRINT_DEBUG("MQTT连接无响应！\n");return Connect_NOTACK;
}/************************************************************************
** 函数名称: MQTT_PingReq								
** 函数功能: 发送MQTT心跳包
** 入口参数: 无
** 出口参数: >=0:发送成功 <0:发送失败
** 备    注: 
************************************************************************/
int32_t MQTT_PingReq(int32_t sock)
{int32_t len;uint8_t buf[200];int32_t buflen = sizeof(buf);	 fd_set readfd;struct timeval tv;tv.tv_sec = 5;tv.tv_usec = 0;FD_ZERO(&readfd);FD_SET(sock,&readfd);			len = MQTTSerialize_pingreq(buf, buflen);transport_sendPacketBuffer(buf, len);//等待可读事件if(select(sock+1,&readfd,NULL,NULL,&tv) == 0)return -1;//有可读事件if(FD_ISSET(sock,&readfd) == 0)return -2;if(MQTTPacket_read(buf, buflen, transport_getdata) != PINGRESP)return -3;return 0;}/************************************************************************
** 函数名称: MQTTSubscribe								
** 函数功能: 订阅消息
** 入口参数: int32_t sock：套接字
**           int8_t *topic：主题
**           enum QoS pos：消息质量
** 出口参数: >=0:发送成功 <0:发送失败
** 备    注: 
************************************************************************/
int32_t MQTTSubscribe(int32_t sock,char *topic,enum QoS pos)
{static uint32_t PacketID = 0;uint16_t packetidbk = 0;int32_t conutbk = 0;uint8_t buf[100];int32_t buflen = sizeof(buf);MQTTString topicString = MQTTString_initializer;  int32_t len;int32_t req_qos,qosbk;fd_set readfd;struct timeval tv;tv.tv_sec = 2;tv.tv_usec = 0;FD_ZERO(&readfd);FD_SET(sock,&readfd);		//复制主题topicString.cstring = (char *)topic;//订阅质量req_qos = pos;//串行化订阅消息len = MQTTSerialize_subscribe(buf, buflen, 0, PacketID++, 1, &topicString, &req_qos);//发送TCP数据if(transport_sendPacketBuffer(buf, len) < 0)return -1;//等待可读事件--等待超时if(select(sock+1,&readfd,NULL,NULL,&tv) == 0)return -2;//有可读事件--没有可读事件if(FD_ISSET(sock,&readfd) == 0)return -3;//等待订阅返回--未收到订阅返回if(MQTTPacket_read(buf, buflen, transport_getdata) != SUBACK)return -4;	//拆订阅回应包if(MQTTDeserialize_suback(&packetidbk,1, &conutbk, &qosbk, buf, buflen) != 1)return -5;//检测返回数据的正确性if((qosbk == 0x80)||(packetidbk != (PacketID-1)))return -6;//订阅成功return 0;
}/************************************************************************
** 函数名称: UserMsgCtl						
** 函数功能: 用户消息处理函数
** 入口参数: MQTT_USER_MSG  *msg：消息结构体指针
** 出口参数: 无
** 备    注: 
************************************************************************/
void UserMsgCtl(MQTT_USER_MSG  *msg)
{//这里处理数据只是打印，用户可以在这里添加自己的处理方式
//   if(msg->msglenth > 2)    //只有当消息长度大于2 "{}" 的时候才去处理它 
//   {PRINT_DEBUG("*****收到订阅的消息！******\n");//返回后处理消息if(msg->msglenth > 2)    //只有当消息长度大于2 "{}" 的时候才去处理它 {switch(msg->msgqos){case 0:PRINT_DEBUG("MQTT>>消息质量：QoS0\n");break;case 1:PRINT_DEBUG("MQTT>>消息质量：QoS1\n");break;case 2:PRINT_DEBUG("MQTT>>消息质量：QoS2\n");break;default:PRINT_DEBUG("MQTT>>错误的消息质量\n");break;}PRINT_DEBUG("MQTT>>消息主题：%s\n",msg->topic);	PRINT_DEBUG("MQTT>>消息类容：%s\n",msg->msg);	PRINT_DEBUG("MQTT>>消息长度：%d\n",msg->msglenth);	 Proscess(msg->msg);}//处理完后销毁数据msg->valid  = 0;
}/************************************************************************
** 函数名称: GetNextPackID						
** 函数功能: 产生下一个数据包ID
** 入口参数: 无
** 出口参数: uint16_t packetid:产生的ID
** 备    注: 
************************************************************************/
uint16_t GetNextPackID(void)
{static uint16_t pubpacketid = 0;return pubpacketid++;
}/************************************************************************
** 函数名称: mqtt_msg_publish						
** 函数功能: 用户推送消息
** 入口参数: MQTT_USER_MSG  *msg：消息结构体指针
** 出口参数: >=0:发送成功 <0:发送失败
** 备    注: 
************************************************************************/
int32_t MQTTMsgPublish(int32_t sock, char *topic, int8_t qos, uint8_t* msg)
{int8_t retained = 0;      //保留标志位uint32_t msg_len;         //数据长度uint8_t buf[MSG_MAX_LEN];int32_t buflen = sizeof(buf),len;MQTTString topicString = MQTTString_initializer;uint16_t packid = 0,packetidbk;//填充主题topicString.cstring = (char *)topic;//填充数据包IDif((qos == QOS1)||(qos == QOS2)){ packid = GetNextPackID();}else{qos = QOS0;retained = 0;packid = 0;}msg_len = strlen((char *)msg_len);//推送消息len = MQTTSerialize_publish(buf, buflen, 0, qos, retained, packid, topicString, (unsigned char*)msg, msg_len);if(len <= 0)return -1;if(transport_sendPacketBuffer(buf, len) < 0)	return -2;	//质量等级0，不需要返回if(qos == QOS0){return 0;}//等级1if(qos == QOS1){//等待PUBACKif(WaitForPacket(sock,PUBACK,5) < 0)return -3;return 1;}//等级2if(qos == QOS2)	{//等待PUBRECif(WaitForPacket(sock,PUBREC,5) < 0)return -3;//发送PUBRELlen = MQTTSerialize_pubrel(buf, buflen,0, packetidbk);if(len == 0)return -4;if(transport_sendPacketBuffer(buf, len) < 0)	return -6;			//等待PUBCOMPif(WaitForPacket(sock,PUBREC,5) < 0)return -7;return 2;}//等级错误return -8;
}/************************************************************************
** 函数名称: ReadPacketTimeout					
** 函数功能: 阻塞读取MQTT数据
** 入口参数: int32_t sock:网络描述符
**           uint8_t *buf:数据缓存区
**           int32_t buflen:缓冲区大小
**           uint32_t timeout:超时时间--0-表示直接查询，没有数据立即返回
** 出口参数: -1：错误,其他--包类型
** 备    注: 
************************************************************************/
int32_t ReadPacketTimeout(int32_t sock,uint8_t *buf,int32_t buflen,uint32_t timeout)
{fd_set readfd;struct timeval tv;if(timeout != 0){tv.tv_sec = timeout;tv.tv_usec = 0;FD_ZERO(&readfd);FD_SET(sock,&readfd); //等待可读事件--等待超时if(select(sock+1,&readfd,NULL,NULL,&tv) == 0)return -1;//有可读事件--没有可读事件if(FD_ISSET(sock,&readfd) == 0)return -1;}//读取TCP/IP事件return MQTTPacket_read(buf, buflen, transport_getdata);
}/************************************************************************
** 函数名称: deliverMessage						
** 函数功能: 接受服务器发来的消息
** 入口参数: MQTTMessage *msg:MQTT消息结构体
**           MQTT_USER_MSG *mqtt_user_msg:用户接受结构体
**           MQTTString  *TopicName:主题
** 出口参数: 无
** 备    注: 
************************************************************************/
void deliverMessage(MQTTString  *TopicName,MQTTMessage *msg,MQTT_USER_MSG *mqtt_user_msg)
{//消息质量mqtt_user_msg->msgqos = msg->qos;//保存消息memcpy(mqtt_user_msg->msg,msg->payload,msg->payloadlen);mqtt_user_msg->msg[msg->payloadlen] = 0;//保存消息长度mqtt_user_msg->msglenth = msg->payloadlen;//消息主题memcpy((char *)mqtt_user_msg->topic,TopicName->lenstring.data,TopicName->lenstring.len);mqtt_user_msg->topic[TopicName->lenstring.len] = 0;//消息IDmqtt_user_msg->packetid = msg->id;//标明消息合法mqtt_user_msg->valid = 1;		
}/************************************************************************
** 函数名称: mqtt_pktype_ctl						
** 函数功能: 根据包类型进行处理
** 入口参数: uint8_t packtype:包类型
** 出口参数: 无
** 备    注: 
************************************************************************/
void mqtt_pktype_ctl(uint8_t packtype,uint8_t *buf,uint32_t buflen)
{MQTTMessage msg;int32_t rc;MQTTString receivedTopic;uint32_t len;switch(packtype){case PUBLISH://拆析PUBLISH消息if(MQTTDeserialize_publish(&msg.dup,(int*)&msg.qos, &msg.retained, &msg.id, &receivedTopic,(unsigned char **)&msg.payload, &msg.payloadlen, buf, buflen) != 1)return;	//接受消息deliverMessage(&receivedTopic,&msg,&mqtt_user_msg);//消息质量不同，处理不同if(msg.qos == QOS0){//QOS0-不需要ACK//直接处理数据UserMsgCtl(&mqtt_user_msg);return;}//发送PUBACK消息if(msg.qos == QOS1){len =MQTTSerialize_puback(buf,buflen,mqtt_user_msg.packetid);if(len == 0)return;//发送返回if(transport_sendPacketBuffer(buf,len)<0)return;	//返回后处理消息UserMsgCtl(&mqtt_user_msg); return;												}//对于质量2,只需要发送PUBREC就可以了if(msg.qos == QOS2){len = MQTTSerialize_ack(buf, buflen, PUBREC, 0, mqtt_user_msg.packetid);			                if(len == 0)return;//发送返回transport_sendPacketBuffer(buf,len);	}		break;case  PUBREL:				           //解析包数据，必须包ID相同才可以rc = MQTTDeserialize_ack(&msg.type,&msg.dup, &msg.id, buf,buflen);if((rc != 1)||(msg.type != PUBREL)||(msg.id != mqtt_user_msg.packetid))return ;//收到PUBREL，需要处理并抛弃数据if(mqtt_user_msg.valid == 1){//返回后处理消息UserMsgCtl(&mqtt_user_msg);}      //串行化PUBCMP消息len = MQTTSerialize_pubcomp(buf,buflen,msg.id);	                   	if(len == 0)return;									//发送返回--PUBCOMPtransport_sendPacketBuffer(buf,len);										break;case   PUBACK://等级1客户端推送数据后，服务器返回break;case   PUBREC://等级2客户端推送数据后，服务器返回break;case   PUBCOMP://等级2客户端推送PUBREL后，服务器返回break;default:break;}
}/************************************************************************
** 函数名称: WaitForPacket					
** 函数功能: 等待特定的数据包
** 入口参数: int32_t sock:网络描述符
**           uint8_t packettype:包类型
**           uint8_t times:等待次数
** 出口参数: >=0:等到了特定的包 <0:没有等到特定的包
** 备    注: 
************************************************************************/
int32_t WaitForPacket(int32_t sock,uint8_t packettype,uint8_t times)
{int32_t type;uint8_t buf[MSG_MAX_LEN];uint8_t n = 0;int32_t buflen = sizeof(buf);do{//读取数据包type = ReadPacketTimeout(sock,buf,buflen,2);if(type != -1)mqtt_pktype_ctl(type,buf,buflen);n++;}while((type != packettype)&&(n < times));//收到期望的包if(type == packettype)return 0;else return -1;		
}void Client_Connect(void)
{char* host_ip;#if  LWIP_DNSip4_addr_t dns_ip;netconn_gethostbyname(HOST_NAME, &dns_ip);host_ip = ip_ntoa(&dns_ip);PRINT_DEBUG("host name : %s , host_ip : %s\n",HOST_NAME,host_ip);
#elsehost_ip = HOST_NAME;
#endif  
MQTT_START: //创建网络连接PRINT_DEBUG("1.开始连接对应云平台的服务器...\n");PRINT_DEBUG("服务器IP地址：%s，端口号：%0d！\n",host_ip,HOST_PORT);while(1){//连接服务器MQTT_Socket = transport_open((int8_t*)host_ip,HOST_PORT);//如果连接服务器成功if(MQTT_Socket >= 0){PRINT_DEBUG("连接云平台服务器成功！\n");break;}PRINT_DEBUG("连接云平台服务器失败，等待3秒再尝试重新连接！\n");//等待3秒vTaskDelay(3000);}PRINT_DEBUG("2.MQTT用户名与秘钥验证登陆...\n");//MQTT用户名与秘钥验证登陆if(MQTT_Connect() != Connect_OK){//重连服务器PRINT_DEBUG("MQTT用户名与秘钥验证登陆失败...\n");//关闭链接transport_close();goto MQTT_START;	 }//订阅消息PRINT_DEBUG("3.开始订阅消息...\n");
//    //订阅消息if(MQTTSubscribe(MQTT_Socket,(char *)TOPIC,QOS1) < 0){//重连服务器PRINT_DEBUG("客户端订阅消息失败...\n");//关闭链接transport_close();goto MQTT_START;	   }	//无限循环PRINT_DEBUG("4.开始循环接收订阅的消息...\n");}/************************************************************************
** 函数名称: mqtt_thread								
** 函数功能: MQTT任务
** 入口参数: void *pvParameters：任务参数
** 出口参数: 无
** 备    注: MQTT连云步骤：
**           1.连接对应云平台的服务器
**           2.MQTT用户与秘钥验证登陆
**           3.订阅指定主题
**           4.等待接收主题的数据与上报主题数据
************************************************************************/
void mqtt_thread(void *pvParameters)
{uint32_t curtick;uint8_t no_mqtt_msg_exchange = 1;uint8_t buf[MSG_MAX_LEN];int32_t buflen = sizeof(buf);int32_t type;fd_set readfd;struct timeval tv;      //等待时间tv.tv_sec = 0;tv.tv_usec = 10;MQTT_START: //开始连接Client_Connect();//获取当前滴答，作为心跳包起始时间curtick = xTaskGetTickCount();while(1){//表明无数据交换no_mqtt_msg_exchange = 1;//推送消息FD_ZERO(&readfd);FD_SET(MQTT_Socket,&readfd);						  //等待可读事件select(MQTT_Socket+1,&readfd,NULL,NULL,&tv);//判断MQTT服务器是否有数据if(FD_ISSET(MQTT_Socket,&readfd) != 0){//读取数据包--注意这里参数为0，不阻塞type = ReadPacketTimeout(MQTT_Socket,buf,buflen,50);if(type != -1){mqtt_pktype_ctl(type,buf,buflen);//表明有数据交换no_mqtt_msg_exchange = 0;//获取当前滴答，作为心跳包起始时间curtick = xTaskGetTickCount();}}//这里主要目的是定时向服务器发送PING保活命令if((xTaskGetTickCount() - curtick) >(KEEPLIVE_TIME/2*1000)){curtick = xTaskGetTickCount();//判断是否有数据交换if(no_mqtt_msg_exchange == 0){//如果有数据交换，这次就不需要发送PING消息continue;}if(MQTT_PingReq(MQTT_Socket) < 0){//重连服务器PRINT_DEBUG("发送保持活性ping失败....\n");goto CLOSE;	 }//心跳成功PRINT_DEBUG("发送保持活性ping作为心跳成功....\n");//表明有数据交换no_mqtt_msg_exchange = 0;}   }CLOSE://关闭链接transport_close();//重新链接服务器goto MQTT_START;	
}void mqtt_send(void *pvParameters)
{int32_t ret;uint8_t no_mqtt_msg_exchange = 1;uint32_t curtick;uint8_t res;/* 定义一个创建信息返回值，默认为pdTRUE */BaseType_t xReturn = pdTRUE;/* 定义一个接收消息的变量 */
//    uint32_t* r_data;	DHT11_Data_TypeDef* recv_data;//初始化json数据cJSON* cJSON_Data = NULL;cJSON_Data = cJSON_Data_Init();double a,b;
MQTT_SEND_START:while(1){xReturn = xQueueReceive( MQTT_Data_Queue,    /* 消息队列的句柄 */&recv_data,      /* 发送的消息内容 */3000); /* 等待时间 3000ms */if(xReturn == pdTRUE){a = recv_data->temperature;b = recv_data->humidity;
//        printf("a = %f,b = %f\n",a,b);//更新数据      res = cJSON_Update(cJSON_Data,TEMP_NUM,&a);res = cJSON_Update(cJSON_Data,HUM_NUM,&b);if(UPDATE_SUCCESS == res){//更新数据成功，char* p = cJSON_Print(cJSON_Data);//发布消息ret = MQTTMsgPublish(MQTT_Socket,(char*)TOPIC,QOS0,(uint8_t*)p);if(ret >= 0){//表明有数据交换no_mqtt_msg_exchange = 0;//获取当前滴答，作为心跳包起始时间curtick = xTaskGetTickCount();				}vPortFree(p);p = NULL;}elsePRINT_DEBUG("update fail\n");}//这里主要目的是定时向服务器发送PING保活命令if((xTaskGetTickCount() - curtick) >(KEEPLIVE_TIME/2*1000)){curtick = xTaskGetTickCount();//判断是否有数据交换if(no_mqtt_msg_exchange == 0){//如果有数据交换，这次就不需要发送PING消息continue;}if(MQTT_PingReq(MQTT_Socket) < 0){//重连服务器PRINT_DEBUG("发送保持活性ping失败....\n");goto MQTT_SEND_CLOSE;	 }//心跳成功PRINT_DEBUG("发送保持活性ping作为心跳成功....\n");//表明有数据交换no_mqtt_msg_exchange = 0;} }
MQTT_SEND_CLOSE://关闭链接transport_close(); //开始连接Client_Connect();goto MQTT_SEND_START;
}void
mqtt_thread_init(void)
{sys_thread_new("iperf_client", mqtt_thread, NULL, 1024, 6);sys_thread_new("iperf_client", mqtt_send, NULL, 1024, 7);
}

4.2.2 方法二

  由于之前的LWIP版本为1.4.1，所以有些包没有。我们可以移植最新的包，lwIP 的项目主页：http://savannah.nongnu.org/projects/lwip/

  主页有以上两个分组：

• Project Homepage：读者可以看到官方对于 lwIP 的说明文档，包括lwIP 更新日记、常见误解、已发现的 BUG、多线程、优化提示和相关文件中的函数描述等内容。
• Domnload Area：读者可以看到 lwIP 源码和 contrib 包的下载网页，如下图所示那样。由于 lwIP 版本居多，因此本教程选择目前最新的 lwIP 版本（2.1.3）。下图中的 contrib 包是提供用户 lwIP 移植文件和 lwIP 相关 demo 例程。注：contrib 包不属于 lwIP内核的一部分，它只是为我们提供移植文件和学习实例。
  

  点击上图中的 lwip-2.1.3.zip 和 contrib-2.1.0.zip 链接，下载完成之后在本地上可以看到这两个压缩包，将lwip\src\apps\mqt路径下的mqtt.c文件添加到工程当中，mqtt.c 文件是lwIP根据MQTT协议规则编写而来的。

4.3 MQTT客户端

1. 文件移植
     这里以阿里云为例，先添加添加hmac_sha和sha1文件，这些文件用来计算核心密钥，这两个文件可在阿里云官方下载

2. 配置阿里云服务器
   ①注册阿里云平台，打开产品分类/物联网 Iot/物联网应用开发，如下图所示
   
   ②点击上图中的“立刻使用”按键进去物联网应用开发页面，在物联网应用开发页面下点击项目管理/新建项目/新建空白项目，在此界面下填写项目名称等相关信息，如下图所示：
   
   ③创建项目完成之后在项目管理页面下点击项目进去子项目管理界面，如下图所示：
   
   ④在上图中点击产品，如下图所示：
   
   ⑤创建产品之后点击图中的设备选项添加设备，如下图所示
   
   ⑥在设备页面下找到我们刚刚创建的设备，如下图所示
   
   ⑦打开“产品/查看/功能定义”路径，在该路径下添加功能定义，如下图所示
   
   ⑧打开自定义功能并发布上线，这里我们添加了两个CurrentTemperature和RelativeHumidity标签

3. 软件设计

  MQTT 配置步骤：

①配置 MCU 为 TCP 客户端模式
②DNS 解析阿里云网页转成 IP 地址
③MQTT 连接
④连接状态
⑤循环发布数据到服务器当中

  程序流程图：

  具体程序以4.2.2方法二为例，在lwip_deom.c文件里实现

#include "FreeRTOS.h"
#include "task.h"
#include "queue.h"
#include <stdint.h>
#include <stdio.h>
#include <netdb.h>
#include "lwip/apps/mqtt.h"
#include "./BSP/LCD/lcd.h"
#include "./MALLOC/malloc.h"
#include "./BSP/LED/led.h"
#include "lwip_demo.h"
#include "hmac.h"
#include "string.h"/* oneNET参考文章：https://open.iot.10086.cn/doc/v5/develop/detail/251 *///static const struct mqtt_connect_client_info_t mqtt_client_info =
//{
//    "MQTT",     /* 设备名 */
//    "366007",   /* 产品ID */
//    "version=2018-10-31&res=products%2F366007%2Fdevices%2FMQTT&et=1672735919&method=md5&sign=qI0pgDJnICGoPdhNi%2BHtfg%3D%3D", /* pass */
//    100,  /* keep alive */
//    NULL, /* will_topic */
//    NULL, /* will_msg */
//    0,    /* will_qos */
//    0     /* will_retain */
//#if LWIP_ALTCP && LWIP_ALTCP_TLS  /* 加密操作，我们一般不使用加密操作 */
//  , NULL
//#endif
//};static ip_addr_t g_mqtt_ip;
static mqtt_client_t* g_mqtt_client;
float g_temp = 0;    /* 温度值 */
float g_humid = 0;   /* 湿度值 */
unsigned char g_payload_out[200];
int g_payload_out_len = 0;
int g_publish_flag = 0;/* 发布成功标志位 */
void lwip_ali_get_password(const char *device_secret, const char *content, char *password);/*** @brief       mqtt进入数据回调函数* @param       arg：传入的参数* @param       data：数据* @param       len：数据大小* @param       flags：标志* @retval      无*/
static void 
mqtt_incoming_data_cb(void *arg, const u8_t *data, u16_t len, u8_t flags)
{const struct mqtt_connect_client_info_t* client_info = (const struct mqtt_connect_client_info_t*)arg;LWIP_UNUSED_ARG(data);printf("\r\nMQTT client \"%s\" data cb: len %d, flags %d\n", client_info->client_id,(int)len, (int)flags);if (flags & MQTT_DATA_FLAG_LAST){/* 从这里可以接收阿里云发布的数据 */ }
}/*** @brief       mqtt进入发布回调函数* @param       arg：传入的参数* @param       topic：主题* @param       tot_len：主题大小* @retval      无*/
static void
mqtt_incoming_publish_cb(void *arg, const char *topic, u32_t tot_len)
{const struct mqtt_connect_client_info_t* client_info = (const struct mqtt_connect_client_info_t*)arg;printf("\r\nMQTT client \"%s\" publish cb: topic %s, len %d\r\n", client_info->client_id,topic, (int)tot_len);
}/*** @brief       mqtt发布回调函数* @param       arg：传入的参数* @param       err：错误值* @retval      无*/
static void
mqtt_publish_request_cb(void *arg, err_t err)
{printf("publish success\r\n");
}/*** @brief       mqtt订阅响应回调函数* @param       arg：传入的参数* @param       err：错误值* @retval      无*/
static void
mqtt_request_cb(void *arg, err_t err)
{const struct mqtt_connect_client_info_t* client_info = (const struct mqtt_connect_client_info_t*)arg;g_publish_flag = 1;printf("\r\nMQTT client \"%s\" request cb: err %d\r\n", client_info->client_id, (int)err);
}/*** @brief       mqtt连接回调函数* @param       client：客户端控制块* @param       arg：传入的参数* @param       status：连接状态* @retval      无*/
static void
mqtt_connection_cb(mqtt_client_t *client, void *arg, mqtt_connection_status_t status)
{err_t err;const struct mqtt_connect_client_info_t* client_info = (const struct mqtt_connect_client_info_t*)arg;LWIP_UNUSED_ARG(client);printf("\r\nMQTT client \"%s\" connection cb: status %d\r\n", client_info->client_id, (int)status);/* 判断是否连接 */ if (status == MQTT_CONNECT_ACCEPTED){/* 判断是否连接 */ if (mqtt_client_is_connected(client)){/* 设置传入发布请求的回调 */mqtt_set_inpub_callback(g_mqtt_client,mqtt_incoming_publish_cb,mqtt_incoming_data_cb,NULL);/* 订阅操作，并设置订阅响应会回调函数mqtt_sub_request_cb */ err = mqtt_subscribe(client, DEVICE_SUBSCRIBE, 1, mqtt_request_cb, arg);if(err == ERR_OK){printf("mqtt_subscribe return: %d\n", err);lcd_show_string(5, 170, 210, 16, 16, "mqtt_subscribe succeed", BLUE);}}}else/* 连接失败 */{printf("mqtt_connection_cb: Disconnected, reason: %d\n", status);} 
}/*** @brief       lwip_demo进程* @param       无* @retval      无*/
void lwip_demo(void)
{struct hostent *server;static struct mqtt_connect_client_info_t mqtt_client_info;server = gethostbyname((char *)HOST_NAME);        /* 对oneNET服务器地址解析 */memcpy(&g_mqtt_ip,server->h_addr,server->h_length); /* 把解析好的地址存放在mqtt_ip变量当中 */char *PASSWORD;PASSWORD = mymalloc(SRAMIN, 300); /* 为密码申请内存 */lwip_ali_get_password(DEVICE_SECRET, CONTENT, PASSWORD);  /*  通过hmac_sha1算法得到password  *//* 设置一个空的客户端信息结构 */memset(&mqtt_client_info, 0, sizeof(mqtt_client_info));/* 设置客户端的信息量 */ mqtt_client_info.client_id = (char *)CLIENT_ID;     /* 设备名称 */mqtt_client_info.client_user = (char *)USER_NAME;   /* 产品ID */mqtt_client_info.client_pass = (char *)PASSWORD;    /* 计算出来的密码 */mqtt_client_info.keep_alive = 100;                  /* 保活时间 */mqtt_client_info.will_msg = NULL;mqtt_client_info.will_qos = NULL;mqtt_client_info.will_retain = 0;mqtt_client_info.will_topic = 0;myfree(SRAMIN, PASSWORD);                 /* 释放内存 *//* 创建MQTT客户端控制块 */g_mqtt_client = mqtt_client_new();/* 连接服务器 */mqtt_client_connect(g_mqtt_client,        /* 服务器控制块 */&g_mqtt_ip, MQTT_PORT,/* 服务器IP与端口号 */mqtt_connection_cb, LWIP_CONST_CAST(void*, &mqtt_client_info),/* 设置服务器连接回调函数 */&mqtt_client_info); /* MQTT连接信息 */while(1){if (g_publish_flag == 1){g_temp = 30 + rand() % 10 + 1;   /* 温度的数据 */g_humid = 54.8 + rand() % 10 + 1;/* 湿度的数据 */sprintf((char *)g_payload_out, "{\"params\":{\"CurrentTemperature\":+%0.1f,\"RelativeHumidity\":%0.1f},\"method\":\"thing.event.property.post\"}", g_temp, g_humid);g_payload_out_len = strlen((char *)g_payload_out);mqtt_publish(g_mqtt_client,DEVICE_PUBLISH,g_payload_out,g_payload_out_len,1,0,mqtt_publish_request_cb,NULL);}vTaskDelay(1000);}
}/*** @brief      将16进制数转化为字符串* @param      pbSrc - 输入16进制数的起始地址* @param      nLen - 16进制数的字节数* @param       pbDest - 存放目标字符串* @retval     无*/
void lwip_ali_hextostr(uint8_t *pbDest, uint8_t *pbSrc, int nLen)
{char ddl, ddh;int i;for (i = 0; i < nLen; i++){ddh = 48 + pbSrc[i] / 16;ddl = 48 + pbSrc[i] % 16;if (ddh > 57) ddh = ddh + 7;if (ddl > 57) ddl = ddl + 7;pbDest[i * 2] = ddh;pbDest[i * 2 + 1] = ddl;}pbDest[nLen * 2] = '\0';
}/*** @brief      通过hmac_sha1算法获取password* @param      device_secret---设备的密钥* @param      content --- 登录密码* @param      password---返回的密码值* @retval     无*/
void lwip_ali_get_password(const char *device_secret, const char *content, char *password)
{char buf[256] = {0};int len = sizeof(buf);hmac_sha1((uint8_t *)device_secret, strlen(device_secret), (uint8_t *)content, strlen(content), (uint8_t *)buf, (unsigned int *)&len);lwip_ali_hextostr((uint8_t *)password, (uint8_t *)buf, len);
}

  在这个文件定义了8个函数，如下所示：

• lwip_demo()：解析域名、配置 mqtt 客户端、连接阿里云服务器的操作
• lwip_ali_hextostr()：将16进制数转化为字符串
• lwip_ali_get_password()：通过hmac_sha1算法获取 password
• mqtt_connection_cb()：mqtt连接回调函数
• mqtt_request_cb()：mqtt订阅响应回调函数
• mqtt_publish_request_cb()：mqtt发布响应回调函数
• mqtt_incoming_publish_cb()：mqtt进入发布回调函数
• mqtt_incoming_data_cb()：mqtt进入数据回调函数