13.6.2 MQTT協(xié)議形象理解

使用電視臺、記者、觀眾三個角色來類比更容易理解：

電視臺：在MQTT里被稱為服務(wù)器（broker）。

有如下作用：

①接受來自客戶的網(wǎng)絡(luò)連接；//記者/觀眾連接進(jìn)電視臺

②接受客戶發(fā)布的應(yīng)用信息；//接受記者發(fā)布的消息

③處理來自客戶端的訂閱和退訂請求；//處理觀眾的訂閱、退訂請求

④向訂閱的客戶轉(zhuǎn)發(fā)應(yīng)用程序消息；//給觀眾轉(zhuǎn)發(fā)記者報(bào)道的消息

記者和觀眾都是客戶端，記者也可以當(dāng)觀眾，觀眾也可以當(dāng)記者。

有如下作用：

①發(fā)布信息；//publish，記者

②訂閱消息；//Subscribe，觀眾

③退訂或刪除消息；

④斷開與服務(wù)器連接

13.6.3 PC MQTT工具使用

1.安裝軟件

軟件在網(wǎng)盤目錄下：

在PC上，需要運(yùn)行MQTT服務(wù)器軟件emqx、MQTT客戶端軟件MQTTX。

對于emqx，把“emqx-5.3.0-windows-amd64.zip”解壓到一個無中文的目錄即可。假設(shè)解壓到了“d:abc”下。

對于MQTTX，雙擊安裝“MQTTX-Setup-1.7.2-x64.exe”即可。

2.啟動與停止MQTT服務(wù)

假設(shè)emqx安裝目錄為“D:abcemqx-5.3.0-windows-amd64”，在命令行執(zhí)行以下命令啟動MQTT服務(wù)：

左右滑動查看完整內(nèi)容

D:abcemqx-5.3.0-windows-amd64inemqx start

然后可通過瀏覽器訪問“http://localhost:18083”以訪問 EMQX Dashboard管理控制臺，進(jìn)行設(shè)備連接與相關(guān)指標(biāo)監(jiān)控管理。默認(rèn)用戶名為“admin”，密碼為“public”

如果要停止MQTT服務(wù)，在命令行下執(zhí)行如下命令：

左右滑動查看完整內(nèi)容

D:abcemqx-5.3.0-windows-amd64inemqx stop

3.使用MQTT客戶端

運(yùn)行MQTTX后，如下圖操作建立連接：

建立連接后，如下圖操作訂閱主題（比如“/topic/humiture”）：

然后，可以自己給自己發(fā)布消息，如下圖操作：

13.6.4 MQTT測試程序

本節(jié)源碼位于如下目錄：

paho.mqtt.c.tar.bz2是庫、示例代碼。

mqtt_test是本節(jié)對應(yīng)的源碼。

測試程序“mqtt_test.c”核心源碼如下：

1.定義一些參數(shù)

左右滑動查看完整內(nèi)容

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

代碼如下：

左右滑動查看完整內(nèi)容

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行，創(chuàng)建一個客戶端。

第71行，設(shè)置回調(diào)函數(shù)：連接斷開時(shí)的回調(diào)函數(shù)、收到消息的回調(diào)函數(shù)、消息發(fā)布成功后的回調(diào)函數(shù)。

第78~81行，設(shè)置用戶名、密碼。

第82行，連接broker。

3.訂閱消息

代碼如下：

左右滑動查看完整內(nèi)容

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 }

當(dāng)接收到訂閱的消息時(shí)，前面第71行設(shè)置的“收到消息的回調(diào)函數(shù)”被調(diào)用，代碼如下：

左右滑動查看完整內(nèi)容

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 }

這個函數(shù)僅僅是把消息打印出來。

4.發(fā)布消息

代碼如下：

左右滑動查看完整內(nèi)容

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行：設(shè)置消息。消息保存在buf里。

第119行：發(fā)布消息。

第125行：等待消息發(fā)送完畢。