玩转PubSubClient MQTT库

时间 2019-11-18

标签 pubsubclient mqtt 繁體版

原文原文链接

1.前言

在ESP8266学习系列中，博主一直使用HTTP协议。HTTP链接属于短链接，而在物联网应用中，普遍应用的倒是MQTT协议。因此，本篇咱们将学习Arduino平台上的MQTT实现库 —— PubSubClient。git

2.MQTT协议

2.1 简介

MQTT协议（Message Queuing Telemetry Transport），翻译过来就是遥信消息队列传输，是IBM公司于1999年提出的，如今最新版本是3.1.1。MQTT是一个基于TCP的发布订阅协议，设计的初始目的是为了极有限的内存设备和网络带宽很低的网络不可靠的通讯，很是适合物联网通讯。数组

MQTT属于应用层协议，基于TCP协议，确保了可靠性。博主在这里不会去详细讲述MQTT协议（网上讲解MQTT协议内容不少，不须要重复），但愿有兴趣的读者自行去阅读，可参考 MQTT中文文档。服务器

MQTT通讯模型以下：网络

发布方（Publisher）将消息发送到 Broker（MQTT服务器）；
Broker 接收到消息之后，检查下都有哪些订阅方订阅了此类消息，而后将消息发送到这些订阅方；
订阅方（Subscriber）从 Broker 获取该消息。

2.2 MQTT消息的QOS

MQTT消息支持三种QOS等级：数据结构

QoS 0：“最多一次”，消息发布彻底依赖底层 TCP/IP 网络。分发的消息可能丢失或重复。例如，这个等级可用于环境传感器数据，单次的数据丢失不要紧，由于不久后还会有第二次发送。
QoS 1：“至少一次”，确保消息能够到达，但消息可能会重复。
QoS 2：“只有一次”，确保消息只到达一次。例如，这个等级可用在一个计费系统中，这里若是消息重复或丢失会致使不正确的收费。

2.3 MQTT控制报文格式

MQTT控制报文由三部分组成：less

固定报头（Fixed header），每一个MQTT控制报文都包含一个固定报头；固定报头指明控制报文类型、标志Flags、剩余长度三大部分。
可变报头（Variable header），某些MQTT控制报文包含一个可变报头部分；它在固定报头和有效载荷之间；可变报头的内容根据报文类型的不一样而不一样，一般包括报文标识符（Packet Identifier）；
有效载荷（Payload），某些MQTT控制报文在报文的最后部分包含一个有效载荷，也就是携带的数据信息；

总体上说，MQTT总体控制报文协议就是：dom

固定报头（必定有） + 可变报头（部分有） + 有效载荷（部分有）ide

数据结构简单，传输数据量小，这也是为何能应用于物联网应用的缘由之一。函数

2.4 MQTT控制报文

2.4.1 CONNECT – 链接服务端

注意点：工具

客户端到服务端的网络链接创建后，客户端发送给服务端的第一个报文必须是CONNECT报文；
在一个网络链接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文看成协议违规处理并断开客户端的链接；
有效载荷包含一个或多个编码的字段。包括客户端的惟一标识符，Will主题，Will消息，用户名和密码。除了客户端标识以外，其它的字段都是可选的，基于标志位来决定可变报头中是否须要包含这些字段。

报文格式：

固定报头 + 可变报头 + 有效载荷

固定报头: MQTTCONNECT(1 << 4)
可变报头: 协议名（Protocol Name），协议级别（Protocol Level），链接标志（Connect Flags）和保持链接（Keep Alive）
有效载荷: 客户端标识符，遗嘱主题，遗嘱消息，用户名，密码

2.4.2 CONNACK —— 确认链接请求

注意点：

服务端发送CONNACK报文响应从客户端收到的CONNECT报文。服务端发送给客户端的第一个报文必须是CONNACK；
若是客户端在合理的时间内没有收到服务端的CONNACK报文，客户端应该关闭网络链接。合理的时间取决于应用的类型和通讯基础设施；

报文格式：

固定报头 + 可变报头

固定报头: MQTTCONNACK(2 << 4)
可变报头: 链接确认标志 + 链接返回码

2.4.3 PUBLISH —— 发布消息

注意点：

PUBLISH控制报文是指从客户端向服务端或者服务端向客户端传输一个应用消息；

报文格式：

固定报头 + 可变报头 + 有效载荷

固定报头: MQTTPUBLISH(3 << 4)，重发标志 DUP，服务质量等级 QoS，保留标志 RETAIN，剩余长度字段
可变报头: 主题名和报文标识符
有效载荷: 被发布的应用消息

2.4.4 PUBACK —— 发布确认

注意点：

PUBACK报文是对QoS 1等级的PUBLISH报文的响应

报文格式：

固定报头 + 可变报头

固定报头: MQTTPUBACK(4 << 4)，剩余长度字段
可变报头: 包含等待确认的PUBLISH报文的报文标识符

2.4.5 PUBREC —— 发布收到（QoS 2，第一步）

注意点：

PUBREC报文是对QoS等级2的PUBLISH报文的响应。它是QoS 2等级协议交换的第二个报文。

报文格式：

固定报头 + 可变报头

固定报头: MQTTPUBREC(5 << 4)，剩余长度字段
可变报头: 包含等待确认的PUBLISH报文的报文标识符

2.4.6 PUBREL —— 发布释放（QoS 2，第二步）

注意点：

PUBREL报文是对PUBREC报文的响应。它是QoS 2等级协议交换的第三个报文。

报文格式：

固定报头 + 可变报头

固定报头: MQTTPUBREL(6 << 4)，剩余长度字段
可变报头: 包含与等待确认的PUBREC报文相同的报文标识符。

2.4.7 PUBCOMP —— 发布完成（QoS 2，第三步）

注意点：

PUBCOMP报文是对PUBREL报文的响应。它是QoS 2等级协议交换的第四个也是最后一个报文。

报文格式：

固定报头 + 可变报头

固定报头: MQTTPUBCOMP(7 << 4)，剩余长度字段
可变报头: 包含与等待确认的PUBREL报文相同的报文标识符

注意点：

客户端向服务端发送SUBSCRIBE报文用于建立一个或多个订阅
为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端
SUBSCRIBE报文也（为每一个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端

报文格式：

固定报头 + 可变报头 + 有效载荷

固定报头: MQTTSUBSCRIBE(8 << 4)，剩余长度字段
可变报头: 报文标识符
有效载荷：包含了一个主题过滤器列表，它们表示客户端想要订阅的主题

2.4.9 SUBACK —— 订阅确认

注意点：

服务端发送SUBACK报文给客户端，用于确认它已收到而且正在处理SUBSCRIBE报文
SUBACK报文包含一个返回码清单，它们指定了SUBSCRIBE请求的每一个订阅被授予的最大QoS等级

报文格式：

固定报头 + 可变报头 + 有效载荷

固定报头: MQTTSUBACK(9 << 4)，剩余长度字段
可变报头: 包含等待确认的SUBSCRIBE报文的报文标识符
有效载荷：包含一个返回码清单。每一个返回码对应等待确认的SUBSCRIBE报文中的一个主题过滤器

2.4.10 UNSUBSCRIBE —— 取消订阅

注意点：

客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题

报文格式：

固定报头 + 可变报头 + 有效载荷

固定报头: MQTTUNSUBSCRIBE(10 << 4)，剩余长度字段
可变报头: 报文标识符
有效载荷：包含客户端想要取消订阅的主题过滤器列表

2.4.11 UNSUBACK —— 取消订阅确认

注意点：

服务端发送UNSUBACK报文给客户端用于确认收到UNSUBSCRIBE报文

报文格式：

固定报头 + 可变报头

固定报头: MQTTUNSUBACK(11 << 4)，剩余长度字段
可变报头: 包含等待确认的UNSUBSCRIBE报文的报文标识符

2.4.12 PINGREQ —— 心跳请求

注意点：

客户端发送PINGREQ报文给服务端
在没有任何其它控制报文从客户端发给服务端时，告知服务端客户端还活着
请求服务端发送响应确认它还活着
使用网络以确认网络链接没有断开
保持链接（Keep Alive）处理中用到这个报文

报文格式：

固定报头

固定报头: MQTTPINGREQ(12 << 4)，剩余长度字段

2.4.13 PINGRESP —— 心跳响应

注意点：

服务端发送PINGRESP报文响应客户端的PINGREQ报文
保持链接（Keep Alive）处理中用到这个报文

报文格式：

固定报头

固定报头: MQTTPINGRESP(13 << 4)，剩余长度字段

2.4.14 DISCONNECT —— 断开链接

注意点：

DISCONNECT报文是客户端发给服务端的最后一个控制报文。
表示客户端正常断开链接。

报文格式：

固定报头

固定报头: MQTTDISCONNECT(14 << 4)，剩余长度字段

3.PubSubClient —— ArduinoMQTT库

老规矩，上一个百度脑图：

3.1 PubSubClient —— 初始化构造器

函数1说明：

/**
 * 建立一个没有初始化的PubSubClient对象
 */
PubSubClient::PubSubClient() {
    this->_state = MQTT_DISCONNECTED;
    this->_client = NULL;
    this->stream = NULL;
    setCallback(NULL);
}

在使用PubSubClient对象以前，必须配置完整的内容：

WiFiClient espClient;
PubSubClient client;

void setup() {
    client.setClient(espClient);
    client.setServer("broker.example.com",1883);
    // client is now configured for use
}

函数2说明：

/**
 * 建立一个部分初始化的PubSubClient对象
 * @param client client实例
 */
PubSubClient::PubSubClient(Client& client) {
    this->_state = MQTT_DISCONNECTED;
    setClient(client);
    this->stream = NULL;
}

在使用PubSubClient对象以前，必须配置完整的内容：

WiFiClient espClient;
PubSubClient client(espClient);

void setup() {
    client.setServer("broker.example.com",1883);
    // client is now configured for use
}

函数3说明：

/**
 * 建立完整初始化的PubSubClient对象
 * @param addr  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 */
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client) {
    this->_state = MQTT_DISCONNECTED;
    setServer(addr, port);
    setClient(client);
    this->stream = NULL;
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param addr  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 * @param stream 输出流，会把收到的消息输出到流中
 */
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, Client& client, Stream& stream) {
    this->_state = MQTT_DISCONNECTED;
    setServer(addr,port);
    setClient(client);
    setStream(stream);
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param addr  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param MQTT_CALLBACK_SIGNATURE callback方法
 * @param client 客户端实例
 */
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
    this->_state = MQTT_DISCONNECTED;
    setServer(addr, port);
    setCallback(callback);
    setClient(client);
    this->stream = NULL;
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param addr  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param MQTT_CALLBACK_SIGNATURE callback方法
 * @param client 客户端实例
 * @param stream 输出流，会把收到的消息输出到流中
 */
PubSubClient::PubSubClient(IPAddress addr, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
    this->_state = MQTT_DISCONNECTED;
    setServer(addr,port);
    setCallback(callback);
    setClient(client);
    setStream(stream);
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param ip  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 */
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client) {
    this->_state = MQTT_DISCONNECTED;
    setServer(ip, port);
    setClient(client);
    this->stream = NULL;
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param ip  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 * @param stream 输出流，会把收到的消息输出到流中
 */
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, Client& client, Stream& stream) {
    this->_state = MQTT_DISCONNECTED;
    setServer(ip,port);
    setClient(client);
    setStream(stream);
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param ip  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 * @param MQTT_CALLBACK_SIGNATURE callback方法
 */
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
    this->_state = MQTT_DISCONNECTED;
    setServer(ip, port);
    setCallback(callback);
    setClient(client);
    this->stream = NULL;
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param ip  mqtt服务器ip地址
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 * @param MQTT_CALLBACK_SIGNATURE callback方法
 * @param stream 输出流，会把收到的消息输出到流中
 */
PubSubClient::PubSubClient(uint8_t *ip, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
    this->_state = MQTT_DISCONNECTED;
    setServer(ip,port);
    setCallback(callback);
    setClient(client);
    setStream(stream);
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param domain  mqtt服务器域名
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 */
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client) {
    this->_state = MQTT_DISCONNECTED;
    setServer(domain,port);
    setClient(client);
    this->stream = NULL;
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param domain  mqtt服务器域名
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 * @param stream 输出流，会把收到的消息输出到流中
 */
PubSubClient::PubSubClient(const char* domain, uint16_t port, Client& client, Stream& stream) {
    this->_state = MQTT_DISCONNECTED;
    setServer(domain,port);
    setClient(client);
    setStream(stream);
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param domain  mqtt服务器域名
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 * @param MQTT_CALLBACK_SIGNATURE callback方法
 */
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client) {
    this->_state = MQTT_DISCONNECTED;
    setServer(domain,port);
    setCallback(callback);
    setClient(client);
    this->stream = NULL;
}

/**
 * 建立完整初始化的PubSubClient对象
 * @param domain  mqtt服务器域名
 * @param post  mqtt服务器端口
 * @param client 客户端实例
 * @param MQTT_CALLBACK_SIGNATURE callback方法
 * @param stream 输出流，会把收到的消息输出到流中
 */
PubSubClient::PubSubClient(const char* domain, uint16_t port, MQTT_CALLBACK_SIGNATURE, Client& client, Stream& stream) {
    this->_state = MQTT_DISCONNECTED;
    setServer(domain,port);
    setCallback(callback);
    setClient(client);
    setStream(stream);
}

3.2 setServer —— 配置服务器

函数说明：

/**
 * 配置服务器
 * @param ip  MQTT服务器ip地址，数组
 * @param port MQTT服务器端口
 */
PubSubClient& PubSubClient::setServer(uint8_t * ip, uint16_t port) {
    IPAddress addr(ip[0],ip[1],ip[2],ip[3]);
    return setServer(addr,port);
}

/**
 * 配置服务器
 * @param ip  MQTT服务器ip地址，IPAddress
 * @param port MQTT服务器端口
 */
PubSubClient& PubSubClient::setServer(IPAddress ip, uint16_t port) {
    this->ip = ip;
    this->port = port;
    this->domain = NULL;
    return *this;
}

/**
 * 配置服务器
 * @param domain  MQTT服务器domain地址
 * @param port MQTT服务器端口
 */
PubSubClient& PubSubClient::setServer(const char * domain, uint16_t port) {
    this->domain = domain;
    this->port = port;
    return *this;
}

注意：

该方法返回this指针，意味着咱们能够实现链式调用；

3.3 setClient —— 配置客户端

函数说明：

/**
 * 配置客户端
 * @param client  client实例，好比wificlient
 */
PubSubClient& PubSubClient::setClient(Client& client){
    this->_client = &client;
    return *this;
}

注意：

该方法返回this指针，意味着咱们能够实现链式调用；

3.4 setStream —— 配置流

函数说明

/**
 * 配置流，可用于存储消息内容
 */
PubSubClient& PubSubClient::setStream(Stream& stream){
    this->stream = &stream;
    return *this;
}

注意：

该方法返回this指针，意味着咱们能够实现链式调用；

3.5 connected —— 判断客户端是否链接上服务器

函数说明

/**
 * 判断client是否链接上服务器
 * @return bool true表示链接上 
 */
boolean PubSubClient::connected() {
    boolean rc;
    if (_client == NULL ) {
        rc = false;
    } else {
        rc = (int)_client->connected();
        if (!rc) {
            //判断链接状态
            if (this->_state == MQTT_CONNECTED) {
                this->_state = MQTT_CONNECTION_LOST;
                _client->flush();
                _client->stop();
            }
        }
    }
    return rc;
}

3.6 connect —— 链接MQTT服务(CONNECT控制报文)

函数说明:

/**
 * 链接MQTT服务(CONNECT控制报文)
 * @param id client端标识符
 * @return bool 是否链接成功
 */
boolean PubSubClient::connect(const char *id) {
    return connect(id,NULL,NULL,0,0,0,0,1);
}

/**
 * 链接MQTT服务(CONNECT控制报文)
 * @param id client端标识符
 * @param user 用户帐号
 * @param pass 用户密码
 * @return bool 是否链接成功 
 */
boolean PubSubClient::connect(const char *id, const char *user, const char *pass) {
    return connect(id,user,pass,0,0,0,0,1);
}

/**
 * 链接MQTT服务(CONNECT控制报文)
 * @param id client端标识符
 * @param willTopic 遗嘱主题
 * @param willQos 遗嘱消息质量等级
 * @param willRetain 是否保留信息
 * @param willMessage 遗嘱内容
 * @return bool 是否链接成功 
 */
boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) {
    return connect(id,NULL,NULL,willTopic,willQos,willRetain,willMessage,1);
}

/**
 * 链接MQTT服务(CONNECT控制报文)
 * @param id client端标识符
 * @param user 用户帐号
 * @param pass 用户密码
 * @param willTopic 遗嘱主题
 * @param willQos 遗嘱消息质量等级
 * @param willRetain 是否保留信息
 * @param willMessage 遗嘱内容
 * @return bool 是否链接成功 
 */
boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage) {
    return connect(id,user,pass,willTopic,willQos,willRetain,willMessage,1);
}

/**
 * 链接MQTT服务(CONNECT控制报文)
 * @param id client端标识符
 * @param user 用户帐号
 * @param pass 用户密码
 * @param willTopic 遗嘱主题
 * @param willQos 遗嘱消息质量等级
 * @param willRetain 是否保留信息
 * @param willMessage 遗嘱内容
 * @param cleanSession 是否清除会话
 * @return bool 是否链接成功 
 *
 * @Note 注意结合CONNECT和CONNACK报文协议
 */
boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage, boolean cleanSession) {
    if (!connected()) {
        int result = 0;

        if (domain != NULL) {
            result = _client->connect(this->domain, this->port);
        } else {
            result = _client->connect(this->ip, this->port);
        }
        if (result == 1) {
            nextMsgId = 1;
            // Leave room in the buffer for header and variable length field
            //固定报头
            uint16_t length = MQTT_MAX_HEADER_SIZE;
            unsigned int j;
//在固定CONNECT报文可变报头包含四个字段，协议名、协议级别、链接标志、保持链接：
#if MQTT_VERSION == MQTT_VERSION_3_1
            uint8_t d[9] = {0x00,0x06,'M','Q','I','s','d','p', MQTT_VERSION};
#define MQTT_HEADER_VERSION_LENGTH 9
#elif MQTT_VERSION == MQTT_VERSION_3_1_1
            uint8_t d[7] = {0x00,0x04,'M','Q','T','T',MQTT_VERSION};
#define MQTT_HEADER_VERSION_LENGTH 7
#endif
            for (j = 0;j<MQTT_HEADER_VERSION_LENGTH;j++) {
                buffer[length++] = d[j];
            }

            /******************** 链接标志（Connect Flags） start *************************/
            uint8_t v;
            //遗嘱主题
            if (willTopic) {
                v = 0x04|(willQos<<3)|(willRetain<<5);
            } else {
                v = 0x00;
            }
            //清除会话
            if (cleanSession) {
                v = v|0x02;
            }
            //帐号密码
            if(user != NULL) {
                v = v|0x80;

                if(pass != NULL) {
                    v = v|(0x80>>1);
                }
            }

            buffer[length++] = v;
           /***********************链接标志（Connect Flags）  end ******************************************************/
           /******************** 保持链接（Keep Alive） start *************************/
            buffer[length++] = ((MQTT_KEEPALIVE) >> 8);
            buffer[length++] = ((MQTT_KEEPALIVE) & 0xFF);
           /******************** 保持链接（Keep Alive） end *************************/

           //有效载荷 CONNECT报文的有效载荷包含一个或多个以长度为前缀的字段，可变报头中的标志决定是否包含这些字段。
           //若是包含的话，必须按照这个顺序出现：client标识符，遗嘱主题，遗嘱消息，用户名，密码。

           /***********  client标识符 start ************/
            CHECK_STRING_LENGTH(length,id)
            length = writeString(id,buffer,length);
           /***********  client标识符 end ************/

           /********* 遗嘱主题  遗嘱消息 start ********/
            if (willTopic) {
                CHECK_STRING_LENGTH(length,willTopic)
                length = writeString(willTopic,buffer,length);
                CHECK_STRING_LENGTH(length,willMessage)
                length = writeString(willMessage,buffer,length);
            }
           /********* 遗嘱主题  遗嘱消息 end ********/

            /********* 用户名  密码 start ********/
            if(user != NULL) {
                CHECK_STRING_LENGTH(length,user)
                length = writeString(user,buffer,length);
                if(pass != NULL) {
                    CHECK_STRING_LENGTH(length,pass)
                    length = writeString(pass,buffer,length);
                }
            }
            /********* 用户名  密码 end ********/

            //拼装 CONNECT消息
            write(MQTTCONNECT,buffer,length-MQTT_MAX_HEADER_SIZE);

            lastInActivity = lastOutActivity = millis();

            //等待MQTT服务器返回响应内容
            while (!_client->available()) {
                unsigned long t = millis();
                //判断是否超时
                if (t-lastInActivity >= ((int32_t) MQTT_SOCKET_TIMEOUT*1000UL)) {
                    _state = MQTT_CONNECTION_TIMEOUT;
                    _client->stop();
                    return false;
                }
            }
            uint8_t llen;
            //读取响应返回的内容
            uint16_t len = readPacket(&llen);

            /*** 处理 CONNACK – 确认链接请求 报文 ***/
            if (len == 4) {
                if (buffer[3] == 0) {
                    lastInActivity = millis();
                    pingOutstanding = false;
                    _state = MQTT_CONNECTED;
                    return true;
                } else {
                    _state = buffer[3];
                }
            }
            _client->stop();
        } else {
            _state = MQTT_CONNECT_FAILED;
        }
        return false;
    }
    return true;
}

注意：

MQTT_KEEPALIVE 默认 15 S；
MQTT_SOCKET_TIMEOUT 默认15 S；

3.7 disconnect —— 断开链接(DISCONNECT报文)

函数说明:

/**
 * 断开链接
 * 客户端断开链接(客户端发给服务端的最后一个控制报文。表示客户端正常断开链接)
 * @Note 取消订阅报文格式: 固定报头(报文类型+剩余长度)
 */
void PubSubClient::disconnect() {
    /*** 断开链接 报文协议 ***/
    buffer[0] = MQTTDISCONNECT;
    buffer[1] = 0;
    _client->write(buffer,2);
    _state = MQTT_DISCONNECTED;
    _client->flush();
    _client->stop();
    lastInActivity = lastOutActivity = millis();
}

函数说明:

/**
 * 订阅主题
 * @param topic 主题
 * @return bool 是否订阅成功
 */
boolean PubSubClient::subscribe(const char* topic) {
    return subscribe(topic, 0);
}

/**
 * 订阅主题(客户端向服务端发送SUBSCRIBE报文用于建立一个或多个订阅)
 * @param topic 主题
 * @param qos 质量等级
 * @return bool 是否订阅成功
 *
 * @Note 订阅报文格式: 固定报头 + 可变报头（报文标识符）+ 有效载荷（主题过滤器）
 *       订阅确认报文格式：固定报头 可变报头（报文标识符）+ 有效载荷（返回码清单）
 */
boolean PubSubClient::subscribe(const char* topic, uint8_t qos) {
    if (qos > 1) {
        return false;
    }
    if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) {
        // Too long
        return false;
    }
    if (connected()) {
        // Leave room in the buffer for header and variable length field
        uint16_t length = MQTT_MAX_HEADER_SIZE;
        nextMsgId++;
        if (nextMsgId == 0) {
            nextMsgId = 1;
        }
        buffer[length++] = (nextMsgId >> 8);
        buffer[length++] = (nextMsgId & 0xFF);
        length = writeString((char*)topic, buffer,length);
        buffer[length++] = qos;
        return write(MQTTSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE);
    }
    return false;
}

3.9 unsubscribe —— 取消订阅主题

函数说明:

/**
 * 取消订阅主题(客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题)
 * @param topic 具体主题
 * @return bool 是否取消成功
 * @Note 取消订阅报文格式: 固定报头(报文类型+剩余长度) + 可变报头（报文标识符）+ 有效载荷（主题过滤器列表）
 */
boolean PubSubClient::unsubscribe(const char* topic) {
    if (MQTT_MAX_PACKET_SIZE < 9 + strlen(topic)) {
        // Too long
        return false;
    }
    if (connected()) {
        uint16_t length = MQTT_MAX_HEADER_SIZE;
        nextMsgId++;
        if (nextMsgId == 0) {
            nextMsgId = 1;
        }
        //可变报头（报文标识符）
        buffer[length++] = (nextMsgId >> 8);
        buffer[length++] = (nextMsgId & 0xFF);
        length = writeString(topic, buffer,length);
        return write(MQTTUNSUBSCRIBE|MQTTQOS1,buffer,length-MQTT_MAX_HEADER_SIZE);
    }
    return false;
}

3.10 publish/publish_P —— 发布消息

函数说明:

/**
 * 发布对应主题消息
 * @param topic 主题
 * @param payload 有效负载
 */
boolean PubSubClient::publish(const char* topic, const char* payload) {
    return publish(topic,(const uint8_t*)payload,strlen(payload),false);
}

/**
 * 发布对应主题消息
 * @param topic 主题
 * @param payload 有效负载
 * @param retained 是否保持
 */
boolean PubSubClient::publish(const char* topic, const char* payload, boolean retained) {
    return publish(topic,(const uint8_t*)payload,strlen(payload),retained);
}

/**
 * 发布对应主题消息
 * @param topic 主题
 * @param payload 有效负载
 * @param plength 负载内容长度
 */
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength) {
    return publish(topic, payload, plength, false);
}

/**
 * 发布对应主题消息
 * @param topic 主题
 * @param payload 有效负载
 * @param plength 负载内容长度
 * @param retained 是否保持
 */
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
    if (connected()) {
        if (MQTT_MAX_PACKET_SIZE < MQTT_MAX_HEADER_SIZE + 2+strlen(topic) + plength) {
            // Too long
            return false;
        }
        // Leave room in the buffer for header and variable length field
        uint16_t length = MQTT_MAX_HEADER_SIZE;
        length = writeString(topic,buffer,length);
        uint16_t i;
        for (i=0;i<plength;i++) {
            buffer[length++] = payload[i];
        }
        uint8_t header = MQTTPUBLISH;
        if (retained) {
            header |= 1;
        }
        return write(header,buffer,length-MQTT_MAX_HEADER_SIZE);
    }
    return false;
}

/**
 * 发布对应主题消息
 * @param topic 主题
 * @param payload 有效负载(F(xxx))
 * @param retained 是否保持
 */
boolean PubSubClient::publish_P(const char* topic, const char* payload, boolean retained) {
    return publish_P(topic, (const uint8_t*)payload, strlen(payload), retained);
}

/**
 * 发布对应主题消息
 * @param topic 主题
 * @param payload 有效负载(F(xxx))
 * @param plength 负载内容长度
 * @param retained 是否保持
 */
boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained) {
    uint8_t llen = 0;
    uint8_t digit;
    unsigned int rc = 0;
    uint16_t tlen;
    unsigned int pos = 0;
    unsigned int i;
    uint8_t header;
    unsigned int len;

    if (!connected()) {
        return false;
    }

    tlen = strlen(topic);

    header = MQTTPUBLISH;
    if (retained) {
        header |= 1;
    }
    buffer[pos++] = header;
    len = plength + 2 + tlen;
    do {
        digit = len % 128;
        len = len / 128;
        if (len > 0) {
            digit |= 0x80;
        }
        buffer[pos++] = digit;
        llen++;
    } while(len>0);

    pos = writeString(topic,buffer,pos);

    rc += _client->write(buffer,pos);

    for (i=0;i<plength;i++) {
        rc += _client->write((char)pgm_read_byte_near(payload + i));
    }

    lastOutActivity = millis();

    return rc == tlen + 4 + plength;
}

3.11 setCallback —— 处理消息回调

函数说明:

/**
 * 设置消息回调函数
 * @param MQTT_CALLBACK_SIGNATURE 
 */
PubSubClient& PubSubClient::setCallback(MQTT_CALLBACK_SIGNATURE) {
    this->callback = callback;
    return *this;
}

注意：

MQTT_CALLBACK_SIGNATURE是一个函数定义

#if defined(ESP8266) || defined(ESP32)
#include <functional>
#define MQTT_CALLBACK_SIGNATURE std::function<void(char*, uint8_t*, unsigned int)> callback
#else
#define MQTT_CALLBACK_SIGNATURE void (*callback)(char*, uint8_t*, unsigned int)
#endif

3.12 loop —— 处理消息以及保持心跳

函数说明:

/**
 * 处理消息以及保持心跳
 */
boolean PubSubClient::loop() {
    if (connected()) {
        unsigned long t = millis();
        if ((t - lastInActivity > MQTT_KEEPALIVE*1000UL) || (t - lastOutActivity > MQTT_KEEPALIVE*1000UL)) {
            if (pingOutstanding) {
                this->_state = MQTT_CONNECTION_TIMEOUT;
                _client->stop();
                return false;
            } else {
                /*** PINGREQ——心跳请求 ****/
                buffer[0] = MQTTPINGREQ;
                buffer[1] = 0;
                _client->write(buffer,2);
                lastOutActivity = t;
                lastInActivity = t;
                pingOutstanding = true;
            }
        }
        if (_client->available()) {
            uint8_t llen;
            uint16_t len = readPacket(&llen);
            uint16_t msgId = 0;
            uint8_t *payload;
            if (len > 0) {
                lastInActivity = t;
                uint8_t type = buffer[0]&0xF0;
                if (type == MQTTPUBLISH) {
                 //服务端发布消息
                    if (callback) {
                        uint16_t tl = (buffer[llen+1]<<8)+buffer[llen+2]; /* topic length in bytes */
                        memmove(buffer+llen+2,buffer+llen+3,tl); /* move topic inside buffer 1 byte to front */
                        buffer[llen+2+tl] = 0; /* end the topic as a 'C' string with \x00 */
                        char *topic = (char*) buffer+llen+2;
                        // msgId only present for QOS>0
                        if ((buffer[0]&0x06) == MQTTQOS1) {
                            msgId = (buffer[llen+3+tl]<<8)+buffer[llen+3+tl+1];
                            payload = buffer+llen+3+tl+2;
                            callback(topic,payload,len-llen-3-tl-2);
                            //客户端发布应答
                            buffer[0] = MQTTPUBACK;
                            buffer[1] = 2;
                            buffer[2] = (msgId >> 8);
                            buffer[3] = (msgId & 0xFF);
                            _client->write(buffer,4);
                            lastOutActivity = t;

                        } else {
                            payload = buffer+llen+3+tl;
                            callback(topic,payload,len-llen-3-tl);
                        }
                    }
                } else if (type == MQTTPINGREQ) {
                    /**** PINGRESP-心跳响应 ****/
                    buffer[0] = MQTTPINGRESP;
                    buffer[1] = 0;
                    _client->write(buffer,2);
                } else if (type == MQTTPINGRESP) {
                    /**** PINGRESP-心跳响应 ****/
                    pingOutstanding = false;
                }
            } else if (!connected()) {
                // readPacket has closed the connection
                return false;
            }
        }
        return true;
    }
    return false;
}

3.13 beginPublish —— 发布大数据(第1步)

上面的发布消息方法都是小数据发布，何为小数据呢？直接看buffer大小吧：

// MQTT_MAX_PACKET_SIZE : Maximum packet size
#ifndef MQTT_MAX_PACKET_SIZE
#define MQTT_MAX_PACKET_SIZE 128
#endif

默认的buffer是128字节，固然博主不推荐你们发送大量数据。
若是要发布稍微大一点的数据，就得用到三部曲方法。

函数说明：

/**
 * 发布大数据(第1步) —— 只是固定报头 + 剩余长度 
 * @param topic 主题
 * @param plength 负载内容长度
 * @param retained 是否保持
 *
 * @Note 这里仍是用到buffer[128]
 */
boolean PubSubClient::beginPublish(const char* topic, unsigned int plength, boolean retained) {
    if (connected()) {
        // Send the header and variable length field
        uint16_t length = MQTT_MAX_HEADER_SIZE;
        length = writeString(topic,buffer,length);
        uint8_t header = MQTTPUBLISH;
        if (retained) {
            header |= 1;
        }
        size_t hlen = buildHeader(header, buffer, plength+length-MQTT_MAX_HEADER_SIZE);
        uint16_t rc = _client->write(buffer+(MQTT_MAX_HEADER_SIZE-hlen),length-(MQTT_MAX_HEADER_SIZE-hlen));
        lastOutActivity = millis();
        return (rc == (length-(MQTT_MAX_HEADER_SIZE-hlen)));
    }
    return false;
}

3.14 write —— 发布大数据(第2步)

函数说明：

/**
 * 发布大数据(第2步) —— 有效负载
 * @param buffer 内容
 * @param size 负载内容长度
 *
 * @Note 这里仍是用到buffer[128]
 */
size_t PubSubClient::write(const uint8_t *buffer, size_t size) {
    lastOutActivity = millis();
    return _client->write(buffer,size);
}

3.15 endPublish —— 发布大数据(第3步)

函数说明：

/**
 * 发布大数据(第3步) 感受没什么用的方法....
 */
int PubSubClient::endPublish() {
 return 1;
}

3.16 state —— 获取Mqtt客户端当前状态

函数说明：

//状态定义
// Possible values for client.state()
#define MQTT_CONNECTION_TIMEOUT     -4
#define MQTT_CONNECTION_LOST        -3
#define MQTT_CONNECT_FAILED         -2
#define MQTT_DISCONNECTED           -1
#define MQTT_CONNECTED               0
#define MQTT_CONNECT_BAD_PROTOCOL    1
#define MQTT_CONNECT_BAD_CLIENT_ID   2
#define MQTT_CONNECT_UNAVAILABLE     3
#define MQTT_CONNECT_BAD_CREDENTIALS 4
#define MQTT_CONNECT_UNAUTHORIZED    5

/**
 * 获取Mqtt客户端当前状态
 */
int PubSubClient::state() {
    return this->_state;
}

4.测试用例

工具讲完了，咱们直接看库自带的示例（后面博主会结合OneNet来使用MQTT 敬请期待）。

4.1 mqtt-8266案例

案例说明：

链接上一个MQTT服务器，每2秒发布一次"hello world"消息到主题“outTopic”
客户端监听主题“inTopic”，并判断负载内容来控制灯亮灭

案例代码：

#include <ESP8266WiFi.h>
#include <PubSubClient.h>

// Update these with values suitable for your network.

const char* ssid = "........";//wifi帐号
const char* password = "........";//wifi秘密
const char* mqtt_server = "broker.mqtt-dashboard.com";//mqtt服务器

WiFiClient espClient;
PubSubClient client(espClient);
long lastMsg = 0;
char msg[50];
int value = 0;

void setup_wifi() {

  delay(10);
  // We start by connecting to a WiFi network
  Serial.println();
  Serial.print("Connecting to ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);

  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }

  randomSeed(micros());

  Serial.println("");
  Serial.println("WiFi connected");
  Serial.println("IP address: ");
  Serial.println(WiFi.localIP());
}

/**
 * 消息回调
 */
void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("Message arrived [");
  Serial.print(topic);
  Serial.print("] ");
  for (int i = 0; i < length; i++) {
    Serial.print((char)payload[i]);
  }
  Serial.println();

  // Switch on the LED if an 1 was received as first character
  if ((char)payload[0] == '1') {
    digitalWrite(BUILTIN_LED, LOW);   // Turn the LED on (Note that LOW is the voltage level
    // but actually the LED is on; this is because
    // it is active low on the ESP-01)
  } else {
    digitalWrite(BUILTIN_LED, HIGH);  // Turn the LED off by making the voltage HIGH
  }

}

/**
 * 断开重连
 */
void reconnect() {
  // Loop until we're reconnected
  while (!client.connected()) {
    Serial.print("Attempting MQTT connection...");
    // Create a random client ID
    String clientId = "ESP8266Client-";
    clientId += String(random(0xffff), HEX);
    // Attempt to connect
    if (client.connect(clientId.c_str())) {
      Serial.println("connected");
      // Once connected, publish an announcement...
      client.publish("outTopic", "hello world");
      // ... and resubscribe
      client.subscribe("inTopic");
    } else {
      Serial.print("failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}

void setup() {
  pinMode(BUILTIN_LED, OUTPUT);     // Initialize the BUILTIN_LED pin as an output
  Serial.begin(115200);
  setup_wifi();
  //配置mqtt服务器地址和端口
  client.setServer(mqtt_server, 1883);
  //设置订阅消息回调
  client.setCallback(callback);
}

void loop() {
  //重连机制
  if (!client.connected()) {
    reconnect();
  }
  //不断监听信息
  client.loop();

  long now = millis();
  if (now - lastMsg > 2000) {
    //每2s发布一次信息
    lastMsg = now;
    ++value;
    snprintf (msg, 50, "hello world #%ld", value);
    Serial.print("Publish message: ");
    Serial.println(msg);
    client.publish("outTopic", msg);
  }
}

4.2 mqtt-auth案例

案例说明：

链接上一个MQTT服务器，须要帐号和密码

案例代码：

#include <ESP8266WiFi.h>
#include <PubSubClient.h>

const char* ssid = "........";
const char* password = "........";
const char* mqtt_server = "broker.mqtt-dashboard.com";

WiFiClient espClient;

void callback(char* topic, byte* payload, unsigned int length) {
  // handle message arrived
}

PubSubClient client(mqtt_server, 1883, callback, espClient);

void setup_wifi() {
  delay(10);
  // We start by connecting to a WiFi network
  Serial.println();
  Serial.print("Connecting to ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);

  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }

  Serial.println("");
  Serial.println("WiFi connected");
  Serial.println("IP address: ");
  Serial.println(WiFi.localIP());
}

void setup()
{
  
  setup_wifi();
  
  // Note - the default maximum packet size is 128 bytes. If the
  // combined length of clientId, username and password exceed this,
  // you will need to increase the value of MQTT_MAX_PACKET_SIZE in
  // PubSubClient.h
  
  if (client.connect("arduinoClient", "testuser", "testpass")) {
    client.publish("outTopic","hello world");
    client.subscribe("inTopic");
  }
}

void loop()
{
  client.loop();
}

4.3 mqtt-larget-message案例

案例说明：

链接上一个MQTT服务器，发布大数据

案例代码：

#include <ESP8266WiFi.h>
#include <PubSubClient.h>

// Update these with values suitable for your network.

const char* ssid = "........";
const char* password = "........";
const char* mqtt_server = "broker.mqtt-dashboard.com";

WiFiClient espClient;
PubSubClient client(espClient);

void setup_wifi() {

  delay(10);
  // We start by connecting to a WiFi network
  Serial.println();
  Serial.print("Connecting to ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);

  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }

  randomSeed(micros());

  Serial.println("");
  Serial.println("WiFi connected");
  Serial.println("IP address: ");
  Serial.println(WiFi.localIP());
}

void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("Message arrived [");
  Serial.print(topic);
  Serial.print("] ");
  for (int i = 0; i < length; i++) {
    Serial.print((char)payload[i]);
  }
  Serial.println();

  // Find out how many bottles we should generate lyrics for
  String topicStr(topic);
  int bottleCount = 0; // assume no bottles unless we correctly parse a value from the topic
  if (topicStr.indexOf('/') >= 0) {
    // The topic includes a '/', we'll try to read the number of bottles from just after that
    topicStr.remove(0, topicStr.indexOf('/')+1);
    // Now see if there's a number of bottles after the '/'
    bottleCount = topicStr.toInt();
  }

  if (bottleCount > 0) {
    // Work out how big our resulting message will be
    int msgLen = 0;
    for (int i = bottleCount; i > 0; i--) {
      String numBottles(i);
      msgLen += 2*numBottles.length();
      if (i == 1) {
        msgLen += 2*String(" green bottle, standing on the wall\n").length();
      } else {
        msgLen += 2*String(" green bottles, standing on the wall\n").length();
      }
      msgLen += String("And if one green bottle should accidentally fall\nThere'll be ").length();
      switch (i) {
      case 1:
        msgLen += String("no green bottles, standing on the wall\n\n").length();
        break;
      case 2:
        msgLen += String("1 green bottle, standing on the wall\n\n").length();
        break;
      default:
        numBottles = i-1;
        msgLen += numBottles.length();
        msgLen += String(" green bottles, standing on the wall\n\n").length();
        break;
      };
    }
  
    // 显示开始发布大数据
    client.beginPublish("greenBottles/lyrics", msgLen, false);
    for (int i = bottleCount; i > 0; i--) {
      for (int j = 0; j < 2; j++) {
        client.print(i);
        if (i == 1) {
          client.print(" green bottle, standing on the wall\n");
        } else {
          client.print(" green bottles, standing on the wall\n");
        }
      }
      client.print("And if one green bottle should accidentally fall\nThere'll be ");
      switch (i) {
      case 1:
        client.print("no green bottles, standing on the wall\n\n");
        break;
      case 2:
        client.print("1 green bottle, standing on the wall\n\n");
        break;
      default:
        client.print(i-1);
        client.print(" green bottles, standing on the wall\n\n");
        break;
      };
    }
    // 发布完毕
    client.endPublish();
  }
}

/**
 * 重连机制
 */
void reconnect() {
  // Loop until we're reconnected
  while (!client.connected()) {
    Serial.print("Attempting MQTT connection...");
    // Create a random client ID
    String clientId = "ESP8266Client-";
    clientId += String(random(0xffff), HEX);
    // Attempt to connect
    if (client.connect(clientId.c_str())) {
      Serial.println("connected");
      // Once connected, publish an announcement...
      client.publish("outTopic", "hello world");
      // ... and resubscribe
      client.subscribe("greenBottles/#");
    } else {
      Serial.print("failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}

void setup() {
  pinMode(BUILTIN_LED, OUTPUT);     // Initialize the BUILTIN_LED pin as an output
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, 1883);
  client.setCallback(callback);
}

void loop() {

  if (!client.connected()) {
    reconnect();
  }
  client.loop();
}

5.总结

整体来讲，MQTT协议简单，很是容易上手使用，但愿读者也能就着协议理解一下源码知识，敬请期待OneNet篇关于MQTT的使用。

玩转PubSubClient MQTT库

1.前言

2.MQTT协议

2.1 简介

2.2 MQTT消息的QOS

2.3 MQTT控制报文格式

2.4 MQTT控制报文

2.4.1 CONNECT – 链接服务端

2.4.2 CONNACK —— 确认链接请求

2.4.3 PUBLISH —— 发布消息

2.4.4 PUBACK —— 发布确认

2.4.5 PUBREC —— 发布收到（QoS 2，第一步）

2.4.6 PUBREL —— 发布释放（QoS 2，第二步）

2.4.7 PUBCOMP —— 发布完成（QoS 2，第三步）

2.4.8 SUBSCRIBE —— 订阅主题

2.4.9 SUBACK —— 订阅确认

2.4.10 UNSUBSCRIBE —— 取消订阅

2.4.11 UNSUBACK —— 取消订阅确认

2.4.12 PINGREQ —— 心跳请求

2.4.13 PINGRESP —— 心跳响应

2.4.14 DISCONNECT —— 断开链接

3.PubSubClient —— ArduinoMQTT库

3.1 PubSubClient —— 初始化构造器

3.2 setServer —— 配置服务器

3.3 setClient —— 配置客户端

3.4 setStream —— 配置流

3.5 connected —— 判断客户端是否链接上服务器

3.6 connect —— 链接MQTT服务(CONNECT控制报文)

3.7 disconnect —— 断开链接(DISCONNECT报文)

3.8 subscribe —— 订阅主题(SUBSCRIBE报文)

3.9 unsubscribe —— 取消订阅主题

3.10 publish/publish_P —— 发布消息

3.11 setCallback —— 处理消息回调

3.12 loop —— 处理消息以及保持心跳

3.13 beginPublish —— 发布大数据(第1步)

3.14 write —— 发布大数据(第2步)

3.15 endPublish —— 发布大数据(第3步)

3.16 state —— 获取Mqtt客户端当前状态

4.测试用例

4.1 mqtt-8266案例

4.2 mqtt-auth案例

4.3 mqtt-larget-message案例

5.总结