OpenHarmony中使用MQTT

作者：張明偉 2023-08-25 09:17:38

? MQTT**(**消息隊(duì)列遙測(cè)傳輸)是ISO 標(biāo)準(zhǔn)(ISO/IEC PRF 20922)下基于發(fā)布/訂閱范式的消息協(xié)議。它工作在TCP/IP協(xié)議族上，是為硬件性能低下的遠(yuǎn)程設(shè)備以及網(wǎng)絡(luò)狀況糟糕的情況下而設(shè)計(jì)的發(fā)布/訂閱型消息協(xié)議，為此，它需要一個(gè)消息中間件。

51CTO 開源基礎(chǔ)軟件社區(qū)

https://ost.51cto.com

一、前景摘要

DES版本：DevEco Studio 3.0 Release
SDK版本：3.2.2.5 （ API9）
npm版本：6.14.16
EMQX：Linux（Ubuntu）
MQTTX：Version: v1.9.2

二、了解MQTT

1、什么是MQTT？

MQTT**(**消息隊(duì)列遙測(cè)傳輸)是ISO 標(biāo)準(zhǔn)(ISO/IEC PRF 20922)下基于發(fā)布/訂閱范式的消息協(xié)議。它工作在TCP/IP協(xié)議族上，是為硬件性能低下的遠(yuǎn)程設(shè)備以及網(wǎng)絡(luò)狀況糟糕的情況下而設(shè)計(jì)的發(fā)布/訂閱型消息協(xié)議，為此，它需要一個(gè)消息中間件。

2、MQTT特性

MQTT協(xié)議是為大量計(jì)算能力有限，且工作在低帶寬、不可靠的網(wǎng)絡(luò)的遠(yuǎn)程傳感器和控制設(shè)備通訊而設(shè)計(jì)的協(xié)議，它具有以下主要的幾項(xiàng)特性：

使用發(fā)布/訂閱消息模式，提供一對(duì)多的消息發(fā)布，解除應(yīng)用程序耦合；

對(duì)負(fù)載內(nèi)容屏蔽的消息傳輸。
使用 TCP/IP 提供網(wǎng)絡(luò)連接。
有三種消息發(fā)布服務(wù)質(zhì)量：QoS（定閱等級(jí)），分0、1、2三個(gè)等級(jí)，簡(jiǎn)單來說是等級(jí)越高越可靠。
小型傳輸，開銷很小（固定長(zhǎng)度的頭部是 2 字節(jié)），協(xié)議交換最小化，以降低網(wǎng)絡(luò)流量。
使用 Last Will 和 Testament 特性通知有關(guān)各方客戶端異常中斷的機(jī)制。

3、應(yīng)用場(chǎng)景（多客戶端，少量消息）

車聯(lián)網(wǎng)
工業(yè)互聯(lián)網(wǎng)
智能家居
視頻直播彈幕
IM實(shí)時(shí)聊天（一對(duì)一聊天，群組聊天）
推送服務(wù)，比如推送實(shí)時(shí)新聞
金融交易數(shù)據(jù)訂閱推送

4、基本名詞

MQTT協(xié)議中的三種身份

MQTT Broker：代理服務(wù)器
Publish：發(fā)布者，發(fā)布消息
Subscribe：訂閱者，訂閱消息

MQTT傳輸?shù)南?/h4>
主題（Topic），負(fù)載（payload）和QoS
Topic：消息的類型，訂閱者訂閱后就會(huì)收到該主題的消息內(nèi)容（payload）.
Payload:消息的內(nèi)容，指訂閱者具體要使用的內(nèi)容。
QoS:服務(wù)質(zhì)量.
“至多一次”（QoS0）：消息發(fā)布完全依賴底層 TCP/IP 網(wǎng)絡(luò)。會(huì)發(fā)生消息丟失或重復(fù)。這一級(jí)別可用于如下情況，環(huán)境傳感器數(shù)據(jù)，丟失一次讀記錄無所謂，因?yàn)椴痪煤筮€會(huì)有第二次發(fā)送。即是推送之后就完事了，至于對(duì)方有沒有收到，收到是什么，數(shù)據(jù)有沒有丟失，都不管。
“至少一次”（QoS1）：確保消息到達(dá)，但消息重復(fù)可能會(huì)發(fā)生。即是你收到推送后，你還得返回一個(gè)puback給對(duì)方，告訴對(duì)方收到了，不然對(duì)方會(huì)以為你沒收到，隔一段時(shí)間后重新給你推送，直到你給對(duì)方返回一個(gè)Puback為止。
“只有一次”(QoS2)：確保消息到達(dá)一次。這一級(jí)別可用于如下情況，在計(jì)費(fèi)系統(tǒng)中，消息重復(fù)或丟失會(huì)導(dǎo)致不正確的結(jié)果。

異常中斷的機(jī)制

使用 Last Will 和 Testament 特性通知有關(guān)各方客戶端。

Last Will：即遺言機(jī)制，用于通知同一主題下的其他設(shè)備發(fā)送遺言的設(shè)備已經(jīng)斷開了連接。
Testament:遺言機(jī)制，功能類似于Last Will。

三、MQTT的簡(jiǎn)單使用

1、搭建MQTT服務(wù)器EMQX

MQTT服務(wù)器有很多種，且部署方式也不一樣。

Linux （Ubuntu）

安裝命令：

curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash

sudo apt-get install emqx ：安裝

sudo systemctl start emqx ：?jiǎn)?dòng)

下載鏈接：https://www.emqx.io/zh/downloads?os=Ubuntu

Docker：https://emqx.io/zh/downloads?os=Docker

Linux（CentOS）：https://www.emqx.io/zh/downloads?os=CentOS

Windows：https://www.emqx.io/zh/downloads?os=Windows

測(cè)試：確保emqx已正常運(yùn)行后，可在Linux本地瀏覽器中輸入： http://127.0.0.1:18083。

注意：未登錄需修改密碼，兩次保持一致。

局域網(wǎng)其他電腦訪問須知Linux的IP地址。

2、MQTTX

創(chuàng)建連接

創(chuàng)建subscription

收發(fā)消息

#代表通配符，代表訂閱所有該類型的topic。

3、在OpenHarmony使用MQTT

安裝依賴

依賴地址：https://gitee.com/openharmony-tpc/ohos_mqtt。

有所不同的是我用的npm安裝的依賴npm install @ohos/mqtt。

創(chuàng)建Mqtt客戶端

創(chuàng)建mqtt客戶端并建立連接：

const clientOptions: MqttClientOptions =  {
  url: '192.168.xxx.xxx/:1883',
  clientId: 'client_id_' + new Date().getTime(),
  persistenceType: 1,
}

 const connectOptions: MqttConnectOptions = {
  userName: '',
  password: '',
  connectTimeout: 30,
}
  this.mqClient = MqttAsync.createMqtt(this.clientOptions);
  this.mqClient.connect(this.connectOptions,  (data: MqttResponse) => {
      console.log(TAG+"  data: "+JSON.stringify(data));
      if (data.code == 0) {
        this.messageArrived();
        this.subscribe('主設(shè)備號(hào)/#');
      }
    });

監(jiān)聽

//接收消息，使用此接口后，當(dāng)訂閱的主題有消息發(fā)布時(shí)，會(huì)自動(dòng)接收到消息。

public messageArrived(): void {
  this.mqClient.messageArrived((err, data) => {
    console.log(TAG+"messageArrived!!!!!!!!!!!");
    console.log(TAG+"messageArrived data:"+JSON.stringify(data));
  });
}

訂閱消息

// 訂閱消息
public subscribe(topic: string, qos: QoS = 1): void {
  const subscribeOption: MqttSubscribeOptions = { topic, qos };
  this.mqClient.subscribe(subscribeOption, (err, data)=>{
      this.handleMessage(data)
  });
}

發(fā)布消息

// 發(fā)布消息
  public publish<T>(topic: string, payload: string | Record<string, any>,  qos: QoS = 0): void {
    if (typeof payload !== 'string') {
      payload = JSON.stringify(payload)
    }
    const payloadLen = payload.length;
    const publishOption: MqttPublishOptions = { topic, payload, qos, payloadLen };
    console.log(TAG, 'publishOption data: ' + JSON.stringify(publishOption));
    this.mqClient.publish(publishOption, (err, data)=>{
       console.log(TAG+"publish!!!!!!!!!!!");
    console.log(TAG+"publish data:"+JSON.stringify(data));
  }));
  }

根據(jù)訂閱的消息的主題的不同進(jìn)行不同的處理

handleMessage(data:any){
     console.log(TAG+"subscribe!!!!!!!!!!!");
    console.log(TAG+"subscribe data:"+JSON.stringify(data));
  //根據(jù)data的不同進(jìn)行不同的處理
}