MQTT开发基础介绍

13.6.2 MQTT协议形象理解

使用电视台、记者、观众三个角色来类比更容易理解：

电视台：在MQTT里被称为服务器（broker）。

有如下作用：

①接受来自客户的网络连接；//记者/观众连接进电视台

②接受客户发布的应用信息；//接受记者发布的消息

③处理来自客户端的订阅和退订请求；//处理观众的订阅、退订请求

④向订阅的客户转发应用程序消息；//给观众转发记者报道的消息

记者和观众都是客户端，记者也可以当观众，观众也可以当记者。

有如下作用：

①发布信息；//publish，记者

②订阅消息；//Subscribe，观众

③退订或删除消息；

④断开与服务器连接

13.6.3 PC MQTT工具使用

1.安装软件

软件在网盘目录下：

在PC上，需要运行MQTT服务器软件emqx、MQTT客户端软件MQTTX。

对于emqx，把“emqx-5.3.0-windows-amd64.zip”解压到一个无中文的目录即可。假设解压到了“d:abc”下。

对于MQTTX，双击安装“MQTTX-Setup-1.7.2-x64.exe”即可。

2.启动与停止MQTT服务

假设emqx安装目录为“D:abcemqx-5.3.0-windows-amd64”，在命令行执行以下命令启动MQTT服务：

左右滑动查看完整内容

D:abcemqx-5.3.0-windows-amd64inemqx start

然后可通过浏览器访问“http://localhost:18083”以访问 EMQX Dashboard管理控制台，进行设备连接与相关指标监控管理。默认用户名为“admin”，密码为“public”

如果要停止MQTT服务，在命令行下执行如下命令：

左右滑动查看完整内容

D:abcemqx-5.3.0-windows-amd64inemqx stop

3.使用MQTT客户端

运行MQTTX后，如下图操作建立连接：

建立连接后，如下图操作订阅主题（比如“/topic/humiture”）：

然后，可以自己给自己发布消息，如下图操作：

13.6.4 MQTT测试程序

本节源码位于如下目录：

paho.mqtt.c.tar.bz2是库、示例代码。

mqtt_test是本节对应的源码。

测试程序“mqtt_test.c”核心源码如下：

1.定义一些参数

左右滑动查看完整内容

22 #define ADDRESS "tcp://192.168.5.10:1883"
23 #define CLIENTID "ExampleClientSub"
24 #define TOPIC_SUBSCRIBE "/iot/down"
25 #define TOPIC_PUBLISH "/iot/up"
26 #define QOS 1
27 #define TIMEOUT 10000L
28
29 #define USER_NAME "100ask"
30 #define PASSWORD "100ask"

2.连接broker

代码如下：

左右滑动查看完整内容

57 int main(int argc, char* argv[])
58 {
59 MQTTClient client;
60 MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
61 int rc;
62
63 if ((rc = MQTTClient_create(&client, ADDRESS, CLIENTID,
64 MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
65 {
66 printf("Failed to create client, return code %d
", rc);
67 rc = EXIT_FAILURE;
68 goto exit;
69 }
70
71 if ((rc = MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivere
d)) != MQTTCLIENT_SUCCESS)
72 {
73 printf("Failed to set callbacks, return code %d
", rc);
74 rc = EXIT_FAILURE;
75 goto destroy_exit;
76 }
77
78 conn_opts.keepAliveInterval = 20;
79 conn_opts.cleansession = 1;
80 conn_opts.username = USER_NAME;
81 conn_opts.password = PASSWORD;
82 if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
83 {
84 printf("Failed to connect, return code %d
", rc);
85 rc = EXIT_FAILURE;
86 goto destroy_exit;
87 }

第63行，创建一个客户端。

第71行，设置回调函数：连接断开时的回调函数、收到消息的回调函数、消息发布成功后的回调函数。

第78~81行，设置用户名、密码。

第82行，连接broker。

3.订阅消息

代码如下：

左右滑动查看完整内容

91 if ((rc = MQTTClient_subscribe(client, TOPIC_SUBSCRIBE, QOS)) != MQTTCLIENT_SU
CCESS)
92 {
93 printf("Failed to subscribe, return code %d
", rc);
94 rc = EXIT_FAILURE;
95 }

当接收到订阅的消息时，前面第71行设置的“收到消息的回调函数”被调用，代码如下：

左右滑动查看完整内容

40 int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *mes
sage)
41 {
42 printf("Message arrived
");
43 printf(" topic: %s
", topicName);
44 printf(" message: %.*s
", message->payloadlen, (char*)message->payload);
45 MQTTClient_freeMessage(&message);
46 MQTTClient_free(topicName);
47 return 1;
48 }

这个函数仅仅是把消息打印出来。

4.发布消息

代码如下：

左右滑动查看完整内容

98 int ch;
99 int cnt = 0;
100 MQTTClient_message pubmsg = MQTTClient_message_initializer;
101 char buf[100];
102 MQTTClient_deliveryToken token;
103
104 while (1)
105 {
106 ch = getchar();
107
108 if (ch =='Q' || ch == 'q')
109 {
110 break;
111 }
112
113 sprintf(buf, "publish %d", cnt++);
114 pubmsg.payload = buf;
115 pubmsg.payloadlen = (int)strlen(buf);
116 pubmsg.qos = QOS;
117 pubmsg.retained = 0;
118
119 if ((rc = MQTTClient_publishMessage(client, TOPIC_PUBLISH, &pubmsg, &t
oken)) != MQTTCLIENT_SUCCESS)
120 {
121 printf("Failed to publish message, return code %d
", rc);
122 continue;
123 }
124
125 rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
126 printf("Message with delivery token %d delivered
", token); 
127 }

第100、113~117行：设置消息。消息保存在buf里。

第119行：发布消息。

第125行：等待消息发送完毕。