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行：等待消息發送完畢。