OpenHarmony中使用MQTT( 二 ) _OpenHarmony

Linux （Ubuntu）安装命令：
curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash
sudo apt-get install emqx ：安装
sudo systemctl start emqx ：启动
下载链接：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
测试：确保emqx已正常运行后，可在Linux本地浏览器中输入： http://127.0.0.1:18083 。

文章插图
注意：未登录需修改密码，两次保持一致。
局域网其他电脑访问须知Linux的IP地址。
2、MQTTX创建连接

文章插图
创建subscription

文章插图
收发消息

文章插图
#代表通配符，代表订阅所有该类型的topic 。
3、在OpenHarmony使用MQTT安装依赖依赖地址：https://gitee.com/openharmony-tpc/ohos_mqtt 。
有所不同的是我用的npm安装的依赖npm install @ohos/mqtt 。
创建Mqtt客户端创建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('主设备号/#');}});

监听//接收消息，使用此接口后，当订阅的主题有消息发布时，会自动接收到消息。

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)});}

发布消息

// 发布消息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));}));}

根据订阅的消息的主题的不同进行不同的处理

handleMessage(data:any){console.log(TAG+"subscribe!!!!!!!!!!!");console.log(TAG+"subscribe data:"+JSON.stringify(data));//根据data的不同进行不同的处理}

四、扩展mqtt与MQ中间件的关系。
消息中间件是基于队列与消息传递技术，在网络环境中为应用系统提供同步或异步、可靠的消息传输的支撑性软件系统。
MQTT 与消息队列有一定的区别，队列是一种先进先出的数据结构，消息队列常用于应用服务层面，实现参考如 RabbitMQ Kafka RocketMQ 。
MQTT 是传输协议，绝大部分 MQTT Broker 不保证消息顺序（Queue），常用语物联网、消息传输等。

【OpenHarmony中使用MQTT】