OpenHarmony开发之MQTT讲解-电子发烧友网

点击蓝字 ╳ 关注我们

开源项目 OpenHarmony

是每个人的 OpenHarmony

王维平

OpenHarmony知识体系工作组

以下内容来自嘉宾分享，不代表开放原子开源基金会观点

相信MQTT这个名称大家都不陌生，物联网的开发必然会遇到MQTT相关知识的应用。那么什么是MQTT？它有什么特点？它能解决什么问题？它是如何工作的？OpenAtom OpenHarmony（以下简称“OpenHarmony”）的物联网开发者要如何使用及验证MQTT功能？接下来的内容将一一为你解答。下图为MQTT通信模型。

什么是MQTT

MQTT（Message Queuing Telemetry Transport消息队列遥测传输）是一种物联网协议，是一种客户端与服务端架构的发布/订阅模式的消息传输协议，旨在为低带宽和不稳定的网络环境中的物联网设备提供可靠的网络服务。MQTT是专门针对物联网开发的轻量级传输协议。MQTT协议针对低带宽网络，低计算能力的设备，做了特殊的优化，使得其能适应各种物联网应用场景。目前它已广泛应用于车联网、智能家居、即时聊天应用和工业互联网等领域。

MQTT的基本原理

在MQTT通讯中，有两个最为重要的角色。它们分别是服务端和客户端。

MQTT服务端

MQTT服务端通常是一台服务器。它是MQTT信息传输的枢纽，负责将MQTT客户端发送来的信息传递给MQTT客户端。MQTT服务端还负责管理MQTT客户端。确保客户端之间的通讯顺畅，保证MQTT消息得以正确接收和准确投递。

MQTT客户端

MQTT客户端可以向服务端发布信息，也可以从服务端收取信息。我们把客户端发送信息的行为称为“发布”信息。而客户端要想从服务端收取信息，则首先要向服务端“订阅”信息。“订阅”信息这一操作很像我们在视频网站订阅某一部电视剧。当这部电视剧上新后，视频网站会向订阅了该剧的用户发送信息，告诉他们有新剧上线了。

MQTT主题

刚刚我们在讲解MQTT客户端订阅信息时，使用了用户在视频网站订阅电视剧这个例子。在MQTT通讯中，客户端所订阅的肯定不是一部部电视剧，而是一个个“主题”。MQTT服务端在管理MQTT信息通讯时，就是使用“主题”来控制的。

为了便于您更好理解服务端是如何通过主题来控制客户端之间的信息通讯，我们来看看下图示例：

以上图示中一共有三个MQTT客户端。它们分别是汽车、手机和电脑。在管理MQTT通讯时，MQTT服务端使用了“主题”来对信息进行管理。比如上图所示，假设我们需要利用手机和电脑获取汽车的速度，那么我们首先要利用电脑和手机向MQTT服务器订阅主题“汽车速度”。接下来，当汽车客户端向服务端的“汽车速度”主题发布信息后，服务端就会首先检查以下都有哪些客户端订阅了“汽车速度”这一主题的信息。当它发现订阅了该主题的客户端有一个手机和一台电脑，于是服务端就会将刚刚收到的“汽车速度”信息转发给订阅了该主题的手机和电脑客户端。

以上实例中，汽车是“汽车速度”主题的发布者，而手机和电脑则是该主题的订阅者。

值得注意的是，MQTT客户端在通讯时，往往角色不是单一的。它既可以作为信息发布者也可以同时作为信息订阅者。如下图所示：

上图中的所有客户端都是围绕“空调温度”这一主题进行通讯的。对于“空调温度”这一主题，手机和电脑客户端成为了MQTT信息的发布者而汽车则成为了MQTT信息的订阅者（接收者）。

（以上讲解参考链接：太极创客http://www.taichi-maker.com/homepage/esp8266-nodemcu-iot/iot-tuttorial/mqtt-tutorial/2-mqtt-basics/）

可以看到，针对不同的主题，MQTT客户端可以切换自己的角色。它们可能对主题A来说是信息发布者，但是对于主题B就成了信息订阅者。

MQTT客户端开发流程

以下采用小熊派的Paho MQTT样例，简要说明MQTT的开发流程。

样例代码在OpenHarmony源码目录/device/board/bearpi/bearpi_hm_nano/app/D5_iot_mqtt，源码下载路径参考文章末尾。开发应用主要涉及以下几个API应用：

MQTT的流程主要由四个步骤组成：

1、创建客户端对象；

2、连接服务器；

3、订阅主题；

4、发布主题。

//订阅的回调函数
void messageArrived(MessageData *data)
{
    printf("Message arrived on topic %.*s: %.*s
", data->topicName->lenstring.len,
        data->topicName->lenstring.data, data->message->payloadlen, data->message->payload);
}
//主流程函数
static void MQTTDemoTask(void)
{
    WifiConnect("BearPi", "123456789");
    printf("Starting ...
");
    int rc, count = 0;
    MQTTClient client;


    NetworkInit(&network);
    printf("NetworkConnect  ...
");


    NetworkConnect(&network, MQTT_SERVERIP, MQTT_SERVERPORT);//本地电脑作为消息代理 此处为电脑IP
    printf("MQTTClientInit  ...
");
    //1-------------创建客户端对象
    MQTTClientInit(&client, &network, MQTT_CMD_TIMEOUT_MS, sendBuf, sizeof(sendBuf), readBuf, sizeof(readBuf));


    MQTTString clientId = MQTTString_initializer;
    clientId.cstring = "bearpi";


    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    data.clientID = clientId;
    data.willFlag = 0;
    data.MQTTVersion = MQTT_VERSION;
    data.keepAliveInterval = MQTT_KEEP_ALIVE_MS;
    data.cleansession = 1;


    printf("MQTTConnect  ...
");
    //2-------------连接服务端
    rc = MQTTConnect(&client, &data);
    if (rc != 0) {
        printf("MQTTConnect: %d
", rc);
        NetworkDisconnect(&network);
        MQTTDisconnect(&client);
        osDelay(MQTT_DELAY_2S);
    }


    printf("MQTTSubscribe  ...
");
    //3-------------订阅主题substopic
    rc = MQTTSubscribe(&client, "substopic", MQTT_QOS, messageArrived);
    if (rc != 0) {
        printf("MQTTSubscribe: %d
", rc);
        osDelay(MQTT_DELAY_2S);
    }
    while (++count) {
        MQTTMessage message;
        char payload[30];


        message.qos = MQTT_QOS;
        message.retained = 0;
        message.payload = payload;
        (void)sprintf_s(payload, sizeof(payload), "message number %d", count);
        message.payloadlen = strlen(payload);
    //4------------发布pubtopic主题
        if ((rc = MQTTPublish(&client, "pubtopic", &message)) != 0) {
            printf("Return code from MQTT publish is %d
", rc);
            NetworkDisconnect(&network);
            MQTTDisconnect(&client);
        }
        osDelay(MQTT_DELAY_500_MS);
    }
}

小熊派开发板MQTT客户端代码一直循环发送主题为pubtopic的信息，信息内容为（"message number %d", count），每次信息count++；

同时开发板客户端也在订阅主题为substopic的信息，一旦接收到substopic信息就会调用回调函数，串口打印出substopic主题的内容。

MQTT实操验证

如何验证MQTT客户端代码是否正常？验证过程主要涉及以下几点：

1、下载消息代理Mosquitto软件，并配置Mosquitto；

2、下载EclipsePahoMQTT工具，并用该工具创建一个客户端，我们简称客户A；

3、修改小熊派客户端MQTT代码相关配置，与第一步配置Mosquitto相匹配，小熊派客户端我们简称客户B。

简要说明下本次验证中涉及的各个模块的作用：

1、消息代理Mosquitto：可以理解为它就是MQTT服务器，所有客户端的消息（发布/订阅）都是与它通信；它负责接收及分发所有信息；

2、EclipsePahoMQTT工具创建的客户端A：我们用来与小熊派创建的客户端B进行信息交互（发布/订阅）。

详细细节：

1、下载消息代理Mosquitto软件，并配置Mosquitto：

（1）点击下载网址（https://mosquitto.org/download/），选择合适的版本，并安装（记录安装路径）；

（2）安装好后，配置Mosquitto，并开启Mosquitto服务：

在Mosquitto软件的安装路径找到mosquitto.conf，打开并作如下修改：

192.168.120.137是本电脑的IP；1883指本次用来验证的服务端口号（本电脑IP192.168.120.137可以有多个服务端口）；allow_anonymous true指允许客户端匿名登录；

修改配置后，在安装目录打开命令窗口，输入.mosquitto -c .mosquitto.conf -v。服务器启动成功后，如下图显示mosquitto version 2.0.11 starting.

2、下载EclipsePahoMQTT工具，创建客户端A，并连接服务器：

3、修改小熊派客户端MQTT代码相关配置，与第一步配置Mosquitto相匹配，小熊派客户端我们简称客户B：

修改连接端代码：

NetworkConnect(&network, 192.168.120.137, 1883);//本地电脑作为消息代理 此处为电脑IP Mosquitto相匹配

4、烧录代码，并操作（发布订阅）通信：

客户端B做了两件事情：1、一直循环发送主题为pubtopic的信息，信息内容是（"message number %d", count）；2、订阅了主题为substopic的信息，一旦服务器有该主题信息就会发送给客户端B，客户端B会把substopic的内容打印。

客户端A也做了两件事：1、订阅主题为pubtopic的信息；2、发布一条主题为substopic的信息，内容为“Hello OpenHarmony!”。

结合客户端B（小熊派开发板）部分代码：

printf("Starting ...
");
NetworkInit(&network);
printf("NetworkConnect  ...
");
NetworkConnect(&network, MQTT_SERVERIP, MQTT_SERVERPORT);//本地电脑作
printf("MQTTClientInit  ...
");
//1-------------创建客户端对象
MQTTClientInit(&client, &network, MQTT_CMD_TIMEOUT_MS, sendBuf, sizeof(sendBuf), readBuf, sizeof(readBuf));


printf("MQTTConnect  ...
");
//2-------------连接服务端
rc = MQTTConnect(&client, &data);
    
printf("MQTTSubscribe  ...
");
//3-------------订阅主题substopic
rc = MQTTSubscribe(&client, "substopic", MQTT_QOS, messageArrived);
(void)sprintf_s(payload, sizeof(payload), "message number %d", count);
 //4------------循环发布pubtopic主题 内容为message number+connt的计数值
 MQTTPublish(&client, "pubtopic", &message)
//订阅的回调函数输出以下内容
printf("Message arrived on topic %.*s: %.*s
", data->topicName->lenstring.len,
data->topicName->lenstring.data,data->message->payloadlen,data->message->payload);