[问答]

MQTT连接的流程是怎样的？

问答对人有帮助，内容完整，我也想知道答案 0 单片机客户端如何应用？ MQTT连接的流程是怎样的？ 0
2021-12-8 07:28:28　　评论淘帖0 邀请回答您可以邀请以下用户，快速回答问题 × heks 该类别下有 50 个回答。邀请回答 hgimtk 该类别下有 44 个回答。邀请回答新星之火12138 该类别下有 42 个回答。邀请回答 wang21cj 该类别下有 41 个回答。邀请回答 chm5 该类别下有 40 个回答。邀请回答 hjfjsdgfjdsf 该类别下有 37 个回答。邀请回答 uvysdfydad 该类别下有 35 个回答。邀请回答 fdjslkjd 该类别下有 35 个回答。邀请回答 werywer 该类别下有 35 个回答。邀请回答 h1654155957.9520 该类别下有 35 个回答。邀请回答 jenny042 该类别下有 34 个回答。邀请回答 dfzvzs 该类别下有 34 个回答。邀请回答维生素B2 该类别下有 34 个回答。邀请回答凤毛麟角该类别下有 34 个回答。邀请回答刘洋该类别下有 33 个回答。邀请回答 meihuacg 该类别下有 33 个回答。邀请回答 hfgdzc 该类别下有 33 个回答。邀请回答 ggfvxv 该类别下有 33 个回答。邀请回答江左盟该类别下有 32 个回答。邀请回答爱与友人该类别下有 32 个回答。邀请回答举报百合子相关推荐 • 请问MQTT协议的整个通信流程是怎样的？ 1450 • MQTT和OTA应用的数据交互流程 3518 • kawayi mqtt连接mosquitto失败怎样解决该问题？ 675 • kawayi mqtt连接mosquitto始终连接不上怎么解决？ 2585 • Android程序是怎样通过SSL单向认证方式连接MQTT服务器的 1388 • 怎么建立MQTT连接？ 2006 • STM32F103板子是如何通过MQTT协议连接阿里云物联网平台的呢 1777 • ESP8266编程以连接到MQTT Broker和Telegram，为什么无法同时连接？ 243 • 怎样在RT-Thread Studio上去开启MQTT示例呢 1447 • 怎样将mqtt移植到单片机工程中呢 708 1个回答

主控+串口透传通信芯片的方式（以STM32+串口WiFi芯片为例,IDE为keil）。
对MQTT不做太深入理解，只需要快速应用即可的人
本文章只针对单片机客户端如何应用进行讲解
对整个MQTT连接的流程进行配合实例讲解。
MQTT 发送和接收均有讲解。
附STM32+透传WiFi芯片源码

如有不合适的读者请止于此，以免浪费时间。

名词解释

[tr]本文中名词本文中的含义[/tr]

客户端	指MQTT客户端
设备	指拥有一定数据处理功能且可以连接网络的功能集合，如STM32+透传WiFi即可以理解为一个设备
账户/密码	指MQTT客户端登录的账户/密码

1、首先先下载MQTT C语言源码GitHub下载源码 C语言

2、建立mqtt工程

将压缩包中
paho.mqtt.embedded-c-master.zippaho.mqtt.embedded-c-masterMQTTPacketsrc 中的全部文件

以及
paho.mqtt.embedded-c-master.zippaho.mqtt.embedded-c-masterMQTTPacketsamples 内的transport.c和transport.h

拷入工程内，最好单独建立一个文件夹，如下图所示。

我们需要做的事情很简单，只需要修改transport.c中的2个函数即可：
int transport_sendPacketBuffer(unsigned char* buf, int buflen);
int transport_getdata(unsigned char* buf, int count);
剩下的这三个可以不用去管，直接清空函数内容即可
int transport_getdatanb(void *sck, unsigned char* buf, int count);
int transport_open(char* host, int port);
int transport_close(int sock);

修改发送函数 transport_sendPacketBuffer(int sock,unsinged char* buf,int buflen)
打开transport.c中可以看到 transport_sendPacketBuffer(int sock,unsinged char* buf,int buflen)的原型如下

int transport_sendPacketBuffer(int sock, unsigned char* buf, int buflen)
{
int rc = 0;
rc = write(sock, buf, buflen);
return rc;
}

我们有必要明白这个函数的作用是什么。这个函数是mqtt协议已经封装好的一个发送函数，目的是将mqtt协议的格式处理函数与硬件接口分开，我们无论采用什么接口，只需要将这个函数改为所采用的硬件接口即可。比如这次我们采用的是串口，那这里我们直接吧这个函数改成串口发送函数：

int transport_sendPacketBuffer( unsigned char* buf, int buflen)
{
WiFi_SendString(buf,buflen);
return 1;
}

其中这里面的串口函数也被我封装了一层，WiFi_SendString(buf,buflen)其实就是一个串口发送函数,我这里用的是串口3：

void WiFi_SendString(uint8_t* str,uint8_t counter)
{
for(int i = 0;i {
USART_SendData(USART3,*str);
while(USART_GetFlagStatus(USART3,USART_FLAG_TXE) == RESET);
str++;
}
}

*需要注意的是，串口发送的数据长度一定要准确，如果在后面多了几个没用的字符，可能会导致服务器接收数据时解析不了*
修改接收函数 int transport_getdata(unsigned char* buf, int count)
该函数在transport.c的原型如下：

int transport_getdata(unsigned char* buf, int count)
{
int rc = recv(mysock, buf, count, 0);
//printf("received %d bytes count %dn", rc, (int)count);
return rc;
}

修改为

int transport_getdata(unsigned char* buf, int count)
{
memcpy(buf,UartFlagStC.UART3String,count);
return count;
}

这个函数的功能很明显，就是将串口收到的数据（UartFlagStC.UART3String）传给入参 buf，传入的长度是count.

如果不懂为什么把这两个函数改写成这样，请接着往下看，如果知道，请直接跳到第三章 [3、MQTT发送]
首先我们看下transport.c中，这两个函数的原型已经给展示出来了

int transport_sendPacketBuffer(int sock, unsigned char* buf, int buflen)
{
  int rc = 0;
  rc = write(sock, buf, buflen);
  return rc;
}

int transport_getdata(unsigned char* buf, int count)
{
  int rc = recv(mysock, buf, count, 0);
  //printf("received %d bytes count %dn", rc, (int)count);
  return rc;
}

其中发送函数调用的是write()，接收调用的是recv()，闹明白这两个函数的功能不就得了？度娘一下：
write()
定义函数
ssize_t write (int fd,const void * buf,size_t count);
函数说明
write()会把指针buf所指的内存写入count个字节到参数fd所指的文件内。当然，文件读写位置也会随之移动。

好了，看到这个解释大家就明白了吧，说白了，transport_sendPacketBuffer()原型就是把长度为count的字符串buf通过socket发送出去，我们用串口，那就通过串口发送就是了！

同理，接收函数调用的是recv(),继续度娘：

recv()
函数原型
int recv( _In_ SOCKET s, _Out_ char *buf, _In_ int len, _In_ int flags);
第一个参数指定接收端套接字描述符；
第二个参数指明一个缓冲区，该缓冲区用来存放recv函数接收到的数据；
第三个参数指明buf的长度；
第四个参数一般置0。

int transport_getdata(unsigned char* buf, int count)的功能大家也应该明白了，这个函数调用recv()，目的就是将接收的socket数据，存到buf里面，长度就是count.我们用的是串口，所以就直接把串口接收到的数据放到buf里面。这里要注意的就是要把当次所有的数据接收完毕后再调用此函数。

3、MQTT发送（整个流程不使用接收函数的情况下）
在进行发送工作之前，我们有必要明白一下MQTT的工作流程，下面以STM32+串口透传模块来讲解一下：

(1)、(2)步首先设置好透传模块。无论是WiFi透传模块还是2/3/4G模块，把该模块连接到mqtt服务器，并进入透传模式即可。
这部分功能需要读者自己去搞定，不同的模块有不同的指令。一般来说这种透传模块一般都是AT+指令，根据手册配置一下即可。这里顺便说一下，MQTT服务器和普通服务器地址差不多，只不过端口号一般是1883而不是8080
（3）连接到服务器之后，客户端需要登录到服务器
登录到服务器可以类比各类门户/社交网站登录账户密码的操作，不废话了，上代码：

char *pubtopic="Topic_Send";//设备发送的主题，服务器订阅的主题
char *subtopic = "Topic_Receive";//设备接收的主题，服务器发送的主题
unsigned char buf[200];
void mqtt_connect(void)
{
u32 len = 0;
MQTTPacket_connectData data = MQTTPacket_connectData_initializer;//配置可变头部分
data.clientID.cstring = "ClientID";//设备唯一ID，请向服务器端工程师索要或者商量制定
data.keepAliveInterval = 60;//保活计时器，即连续两次心跳包发送的最大时间间隔（单位：秒）
data.cleansession = 1;//1：丢弃之前的连接信息
data.username.cstring = "userName" ;//用户名，向服务器端工程师索要即可
data.password.cstring = "password";//密码，向服务器端工程师索要即可
data.MQTTVersion = 4;
len = MQTTSerialize_connect(buf, buflen, &data); //构建连接报文并获取长度
transport_sendPacketBuffer(buf,len);//通过硬件接口发送给服务器已经封装好的报文
while(!UartFlagStC.MQTTReceiveFlag){};//等待连接成功----------此处下文会着重讲述  注释（1）
UartFlagStC.MQTTReceiveFlag =0;
}

基本上上述代码不需要进行改动，只需要将data结构体将需要的信息写进去即可。

注释1：在最后一行代码里面，有一个while（），目的是循环直至登录服务器成功。当登录服务器成功之后，服务器会发给设备一条数据，此时即表示登录成功。简单的方法是采用的是在串口中断接收函数里设置一个标志位UartFlagStC.MQTTReceiveFlag，只要串口有数据接收，将该标志为置1，即默认连接成功（本来模块已经进入透传模式，只要有数据肯定是服务器发来的，只要服务器发来数据，那么肯定可以判定登录成功）。
标准接收数据处理方法将在 [4、MQTT接收数据] 中详细讲解;
简单方法实现如下：

/***************串口中断函数******************/
void USART3_IRQHandler(void)
{
if(USART_GetITStatus(USART3, USART_IT_RXNE) != RESET)
{
      if(UartFlagStC.UART3Counter<200)
      {
      UartFlagStC.UART3Flag = UART3_HANDLE;
      UartFlagStC.UART3String[UartFlagStC.UART3Counter] = USART_ReceiveData(USART3);
      UartFlagStC.UART3Counter++;
      }else
{
UartFlagStC.UART3Counter = 0;
memset(UartFlagStC.UART3String,0,200);
}
UartFlagStC.MQTTReceiveFlag =1;//将该标志为置1
USART_ClearITPendingBit(USART3,USART_IT_RXNE);
}
}

(4)心跳包
心跳包的意义可以从名称上看出来，主要作用就是和服务器保持一个长连接，每隔一段时间发给服务器一个数据包，告诉服务器–“我“还在线上。
心跳包发送的最大间隔与这一句代码有关：
data.keepAliveInterval = 60;//保活计时器，即连续两次心跳包发送的最大时间间隔（单位：秒）
1
即两次心跳包发送时间不得大于这个设置的数值，如大于这个数值的1.5倍时间后还未收到心跳包，服务器会将该客户端踢下线（如上述设置时间为60S，如果过了60 ×1.5=90S后仍未收到心跳包，则判定设备离线，服务器会将该设备踢下线）
发送心跳包的方法：

void MQTTSendHeartbeat(void)
{
char h_buf[200] = {0};//长度自行设定测试
u8 len = MQTTSerialize_pingreq(h_buf, 200);//调用mqtt心跳包封装函数
transport_sendPacketBuffer((unsigned char*)h_buf, len);//发送数据
}

建议心跳包的发送在主循环或者定时器中断函数中进行定时循环发送。
（5）数据发送
先上代码：
int mqtt_publish(char *pTopic,char *pMessage,u8 mslen)
{
char buf[200] = {0};//长度自行设定测试
MQTTString topicString = MQTTString_initializer;
int msglen = mslen;//
int buflen = sizeof(buf);
memset(buf,0,buflen);
topicString.cstring = pTopic;//
int32_t  len = MQTTSerialize_publish(buf, buflen, 0, 0, 0, 0, topicString, (unsigned char*)pMessage, msglen); //
transport_sendPacketBuffer(buf,len);//
return 0;
}

函数讲解：
入参：char *pTopic : 为服务器订阅主题，设备往这个主题发送数据，服务器即可收到；当然前提是登录成功之后；
剩下俩个就是要发送的内容和长度。
每当代码调用此函数，就会向指定主题发送数据。

（6）发送总结
简单地总结下发送数据我们所需做的工作：
（1）将WiFi芯片（或者其他无线芯片），连接至路由器，连接MQTT服务器（网址和端口号问你们服务器端。。。），并进入透传模式；
（2）之后登录mqtt服务器，根据上文，就是调用这个函数：void mqtt_connect(void);
（3）然后再循环发送心跳包。
简单写一个流程：
//伪代码
int main(void）
{
WiFi_Config();//连接路由&&连接服务器&&进入透传
mqtt_connect();//登录服务器
while(1)
{
delay_ms(XX);//具体延时多久根据你设定的data.keepAliveInterval值来定，设定值必须小于keepAliveInterval哦！
MQTTSendHeartbeat();//发送心跳包
mqtt_publish(pubtopic,data,data_len);//给服务器发送数据，想在哪发在哪发，这里举个栗子
}
}

4、MQTT接收
MQTT 消息收发模式可以分为两种：Pub/Sub和P2P模式，简单地理解为群发/点对点模式。详细信息可以根据阿里文档了解下两种模式的区别

1、收发模式讲解
（1）Pub/Sub模式：
如果读者仔细看过上面的内容，可以发现

char *pubtopic="Topic_Send";//设备发送的主题，服务器订阅的主题
char *subtopic = "Topic_Receive";//设备接收的主题，服务器发送的主题
1
2
中的subtopic并未使用，此项内容即为设备订阅的主题，读者在实际使用的时候需要自行更改为自己使用的主题；

客户端主题订阅：

int mqtt_subscribe(char *pTopic)
{
int32_t len;
MQTTString topicString = MQTTString_initializer;
int buflen = sizeof(buf);
memset(buf,0,buflen);
topicString.cstring = pTopic;//
len = MQTTSerialize_subscribe(buf, buflen, 0, 1, 1, &topicString, 0);
transport_sendPacketBuffer((unsigned char*)buf, len);
return 0;
}

调用此函数，即订阅该subtopic 主题。每当服务器给这个主题发送数据，订阅该主题的设备都会收到改数据。

（2）P2P模式：
点对点模式，那设备就没什么事情可做了，不需要订阅主题，只用登录到服务器等着数据过来解析就好了。服务器主要做的就是将数据发送给想法送的对象，根据data.clientID.cstring即设备的clientID来判定。

2、数据解析
简单地方法：
串口收到的数据是服务器封装后的，我们可以不用管mqtt协议的包头包尾，直接在数据中定义一套包头包尾，我们在解析的时候直接把mqtt协议的包头包尾过滤掉，找到我们自己的协议进而解析数据。我这边举个栗子：

u8 strhangdle[200] = {0};
u8 flag = 0;
u8 i = 0;
for(i = 0;i {
if(flag == 0x01)
{
if(UartFlagStC.UART3String == 0xaa)
{
flag |= 0x02;
}
else flag = 0;
}
if(flag == 0x03)
{
break;
}
if((UartFlagStC.UART3String == 0xaa)&&(flag == 0))
{
flag = 0x01;
}
}
memcpy(strhangdle,&UartFlagStC.UART3String[i-1],UartFlagStC.UART3Counter-i+1);
if((strhangdle[0] == 0xaa)&&(strhangdle[1] == 0xaa))
{
WiFiProtocolHandle(strhangdle);//处理协议
for(uint8_t i =0;i UartFlagStC.UART3String = 0;
UartFlagStC.UART3Counter = 0;
UartFlagStC.UART3Flag =0;
return;
}

我这里的包头是0XAA-0XAA，当收到这两个数据之后则表示是服务器给设备发来的数据，我剩下来只需要解析这些数据，即可完成基本任务。当然这是一种简单地方法，有些鸡肋，但还是可以快速解决问题。

标准方法
既然我们改写了transport_getdata(unsigned char* buf, int count)，那不用这个函数好像是太浪费了，接收的标准方法先上代码：

/*******************
需要说明的是MQTTPacket_read(buf, UartFlagStC.UART3Counter, transport_getdata)入参中的
UartFlagStC.UART3Counter，是串口收到数据的总长度。
********************/
char buf[200] = {0};
void GetDataHandle(void)
{
switch(MQTTPacket_read(buf, UartFlagStC.UART3Counter, transport_getdata))
{
case CONNECT:
/******添加处理函数*****/
break;
case CONNACK:
/******添加处理函数*****/
break;
case PUBLISH:
/******添加处理函数*****/
break；
...
case DISCONNECT:
/******添加处理函数*****/
break;
}
}

需要说明的是MQTTPacket_read(buf, UartFlagStC.UART3Counter, transport_getdata)入参中的
UartFlagStC.UART3Counter，是串口收到数据的总长度。

上面的代码就是根据接收到的数据用mqtt协议进行解析，调用MQTTPacket_read（）进行解析数据，根据服务器发来的不同类型的数据，进行不同的解析，比如PUBLISH是服务器发来的设备订阅/P2P的数据，我们就可以根据改数据进行回复或者控制设备。
函数原型为
int MQTTPacket_read(unsigned char* buf, int buflen, int (*getfn)(unsigned char*, int))
可以看出，主要就是一个函数指针，前面的buf和buflen，就是传递给该函数指针的，可以看出，该函数指针入参类型和返回值，与我们已经更改完毕的int transport_getdata(unsigned char* buf, int count)是一致的。
至于MQTTPacket_read()的返回值，在MQTTPacket.c里面有一个枚举类型的数据，我这里拿出来给大家看一下：

enum msgTypes
{
CONNECT = 1, CONNACK, PUBLISH, PUBACK, PUBREC, PUBREL,
PUBCOMP, SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK,
PINGREQ, PINGRESP, DISCONNECT
};

不同类型的信息代表含义如下：

[tr]类型解释[/tr]
1 CONNECT – 连接服务端：客户端到服务端的网络连接建立后，客户端发送给服务端的第一个报文必须是CONNECT报文
2 CONNACK – 确认连接请求：服务端发送CONNACK报文响应从客户端收到的CONNECT报文。服务端发送给客户端的第一个报文必须是CONNACK。如果客户端在合理的时间内没有收到服务端的CONNACK报文，客户端应该关闭网络连接。合理的时间取决于应用的类型和通信基础设施。
3 PUBLISH – 发布消息：PUBLISH控制报文是指从客户端向服务端或者服务端向客户端传输一个应用消息。
4 PUBACK –发布确认：PUBACK报文是对QoS 1等级的PUBLISH报文的响应。
5 PUBREC – 发布收到（ QoS 2，第一步）：PUBREC报文是对QoS等级2的PUBLISH报文的响应。它是QoS 2等级协议交换的第二个报文。
6 PUBREL – 发布释放（ QoS 2，第二步）：PUBREL报文是对PUBREC报文的响应。它是QoS 2等级协议交换的第三个报文。
7 PUBCOMP – 发布完成（ QoS 2，第三步）：PUBCOMP报文是对PUBREL报文的响应。它是QoS 2等级协议交换的第四个也是最后一个报文。
8 SUBSCRIBE - 订阅主题：客户端向服务端发送SUBSCRIBE报文用于创建一个或多个订阅。每个订阅注册客户端关心的一个或多个主题。为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端。 SUBSCRIBE报文也（为每个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端。
9 SUBACK – 订阅确认：服务端发送SUBACK报文给客户端，用于确认它已收到并且正在处理SUBSCRIBE报文。
10 UNSUBSCRIBE –取消订阅：客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题。
11 UNSUBACK – 取消订阅确认：服务端发送UNSUBACK报文给客户端用于确认收到UNSUBSCRIBE报文。
12 PINGREQ – 心跳请求：客户端发送PINGREQ报文给服务端的。用于：1. 在没有任何其它控制报文从客户端发给服务的时，告知服务端客户端还活着。2. 请求服务端发送响应确认它还活着。3. 使用网络以确认网络连接没有断开。
13 PINGRESP – 心跳响应：服务端发送PINGRESP报文响应客户端的PINGREQ报文。表示服务端还活着。
14 DISCONNECT –断开连接：DISCONNECT报文是客户端发给服务端的最后一个控制报文。表示客户端正常断开连接。
需要注意的是，所有类型不一定都需要判断，只需要选取用到的一些即可。
以上就是MQTT单片机设备端的快速应用，下面是代码中需要注意和修改的地方：

这个是我模块连接到路由器后需要连接服务器的步骤，这里自己下载调试的时候改成自己的服务器地址。

这里 clientID、用户名、密码自己的填进去就可以了。这里在多说一句，每台机器的clientID要独一无二，如果有重复且data.cleansession置1的话，那么就会导致后登陆的会踢掉前面的设备，引起一些数据丢失或者设备不受控制等状况。