[[353642]]

想了解更多內(nèi)容，請?jiān)L問：

51CTO和華為官方合作共建的鴻蒙技術(shù)社區(qū)

https://harmonyos.51cto.com/#zz

我們使用的是paho mqtt軟件包，這里介紹一下怎么使用mqtt協(xié)議編程。關(guān)于鴻蒙系統(tǒng)的mqtt移植好的軟件包，相關(guān)github鏈接如下：

https://gitee.com/qidiyun/harmony_mqtt

這里提供一個(gè)簡單的編程示例：

這里我們使用MQTTClient編程模型，他支持多任務(wù)多線程，非常適合用在鴻蒙系統(tǒng)上。

1. 網(wǎng)絡(luò)初始化

這里定義一個(gè) Network 結(jié)構(gòu)體，然后指定我們的MQTT服務(wù)器的IP和端口號。

Network n; 
   //初始化結(jié)構(gòu)體 
 NetworkInit(&n); 
 //連接到指定的MQTT服務(wù)器IP、端口號 
 NetworkConnect(&n, “XXX.XXX.XXX.XXX”, XXXX); 

 2. 設(shè)置MQTT緩存和啟動(dòng)MQTT線程

我們這里使用的是MQTT線程功能。

MQTTClientInit(&c, &n, 1000, buf, 100, readbuf, 100); 
 MQTTStartTask(&c); 

 3. 設(shè)置MQTT相關(guān)參數(shù)

接下來我們設(shè)置MQTT的相關(guān)參數(shù)，包括版本號、客戶端ID、賬戶密碼等

MQTTPacket_connectData data = MQTTPacket_connectData_initializer;  
 data.willFlag = 0; 
 //MQTT版本為 v3 
 data.MQTTVersion = 3; 
 //設(shè)置客戶端ID 
 data.clientID.cstring = opts.clientid; 
 //設(shè)置客戶端賬戶 
 data.username.cstring = opts.username; 
 //設(shè)置客戶端密碼 
 data.password.cstring = opts.password; 
 data.keepAliveInterval = 10; 
 data.cleansession = 1; 
 
 //連接到MQTT服務(wù)器 
 rc = MQTTConnect(&c, &data); 

 4. 訂閱主題和接收消息

訂閱主題可以使用如下函數(shù)

MQTTSubscribe(&c, topic, opts.qos, messageArrived); 

它的函數(shù)原型如下：

DLLExport int MQTTSubscribe(MQTTClient* client, const char* topicFilter, enum QoS, messageHandler); 

其中：

void messageArrived(MessageData* md) 
{ 
 MQTTMessage* message = md->message; 
 //打印接收到的消息的長度、和消息內(nèi)容 
 printf("%.*s", (int)message->payloadlen, (char*)message->payload); 
} 

 5. 發(fā)送消息

發(fā)送消息也比較簡單，我們只需要設(shè)置好我們的主題和消息內(nèi)容即可

memset(&pubmsg, '\0', sizeof(pubmsg)); 
 //消息內(nèi)容為 hello harmonyOS ! 
   pubmsg.payload = (void*)"hello harmonyOS !"; 
 //消息長度 
   pubmsg.payloadlen = strlen((char*)pubmsg.payload); 
   pubmsg.qos = QOS0; 
   pubmsg.retained = 0; 
   pubmsg.dup = 0; 
 
 //推送消息，主題為 pubtest 
 MQTTPublish(&c, "pubtest", &pubmsg); 

 完整源碼如下： 

#include <stdio.h> 
 
#include <unistd.h> 
 
#include "ohos_init.h" 
#include "cmsis_os2.h" 
 
#include <unistd.h> 
#include "hi_wifi_api.h" 
//#include "wifi_sta.h" 
#include "lwip/ip_addr.h" 
#include "lwip/netifapi.h" 
 
#include "lwip/sockets.h" 
 
#include "MQTTClient.h" 
 
/** 
 * MQTT URI farmat: 
 * domain mode 
 * tcp://iot.eclipse.org:1883 
 * 
 * ipv4 mode 
 * tcp://192.168.10.1:1883 
 * ssl://192.168.10.1:1884 
 * 
 * ipv6 mode 
 * tcp://[fe80::20c:29ff:fe9a:a07e]:1883 
 * ssl://[fe80::20c:29ff:fe9a:a07e]:1884 
 */ 
#define MQTT_URI                "tcp://106.13.62.194:1883" 
 
struct opts_struct 
{ 
    char* clientid; 
    int nodelimiter; 
    char* delimiter; 
    enum QoS qos; 
    char* username; 
    char* password; 
    char* host; 
    int port; 
    int showtopics; 
} opts = 
{ 
    (char*)"stdout-subscriber", 0, (char*)"\n", QOS2, NULL, NULL, (char*)"106.13.62.194", 1883, 1 
}; 
 
 
void messageArrived(MessageData* md) 
{ 
    MQTTMessage* message = md->message; 
 
    if (opts.showtopics) 
        printf("%.*s\t", md->topicName->lenstring.len, md->topicName->lenstring.data); 
    if (opts.nodelimiter) 
        printf("%.*s", (int)message->payloadlen, (char*)message->payload); 
    else 
        printf("%.*s%s", (int)message->payloadlen, (char*)message->payload, opts.delimiter); 
    //fflush(stdout); 
} 
 
 
unsigned char buf[100]; 
unsigned char readbuf[100]; 
 
int mqtt_test(void) 
{ 
    int rc = 0; 
     
    MQTTMessage pubmsg; 
 
     
    char* topic = "test"; 
 
    if (strchr(topic, '#') || strchr(topic, '+')) 
        opts.showtopics = 1; 
    if (opts.showtopics) 
        printf("topic is %s\n", topic); 
 
    Network n; 
    MQTTClient c; 
 
    NetworkInit(&n); 
    NetworkConnect(&n, opts.host, opts.port); 
     
    MQTTClientInit(&c, &n, 1000, buf, 100, readbuf, 100); 
    MQTTStartTask(&c); 
 
    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;        
    data.willFlag = 0; 
    data.MQTTVersion = 3; 
    data.clientID.cstring = opts.clientid; 
    data.username.cstring = opts.username; 
    data.password.cstring = opts.password; 
 
    data.keepAliveInterval = 10; 
    data.cleansession = 1; 
    printf("Connecting to %s %d\n", opts.host, opts.port); 
     
    rc = MQTTConnect(&c, &data); 
    printf("Connected %d\n", rc); 
     
     
 
    printf("Subscribing to %s\n", topic); 
    rc = MQTTSubscribe(&c, topic, opts.qos, messageArrived); 
    printf("Subscribed %d\n", rc); 
 
    memset(&pubmsg, '\0', sizeof(pubmsg)); 
    pubmsg.payload = (void*)"hello harmonyOS !"; 
    pubmsg.payloadlen = strlen((char*)pubmsg.payload); 
    pubmsg.qos = QOS0; 
    pubmsg.retained = 0; 
    pubmsg.dup = 0; 
 
     
 
    while (1) 
    { 
        MQTTPublish(&c, "pubtest", &pubmsg); 
        sleep(1);    
    } 
     
    printf("Stopping\n"); 
 
    MQTTDisconnect(&c); 
    NetworkDisconnect(&n); 
 
    return 0; 
} 

 想了解更多內(nèi)容，請?jiān)L問：

51CTO和華為官方合作共建的鴻蒙技術(shù)社區(qū)

https://harmonyos.51cto.com/#zz