HarmonyOS实现MQTT消息监控和展示_0

JS代码：constABILITY_TYPE_EXTERNAL=0;constACTION_SYNC=0;constACTION_MESSAGE_CODE_START_MQTT=1001;constACTION_MESSAGE_CODE_MQTT_MESSAGE=1002;constBUNDLE_NAME='com.example.mqttaapplication';constABILITY_NAME='com.example.mqttaapplication.PlayAbilityAbility';exportconst={startMqtt:asyncfunction(){FeatureAbility.callAbility({messageCode:ACTION_MESSAGE_CODE_START_MQTT,abilityType:ABILITY_TYPE_EXTERNAL,syncOption:ACTION_SYNC,bundleName:BUNDLE_NAME,abilityName:ABILITY_NAME});},mqttMessage:asyncfunction(that){varresult=awaitatureAbility.callAbility({messageCode:ACTION_MESSAGE_CODE_MQTT_MESSAGE,abilityType:ABILITY_TYPE_EXTERNAL,syncOption:ACTION_SYNC,bundleName:BUNDLE_NAME,abilityName:ABILITY_NAME});varret=JSON.parse(result);if(ret.code==0){console.info('mqttis:'+JSON.stringify(ret.abilityResult));that.title='mqttis:'+JSON.stringify(ret.abilityResult);}else{console.error('mqtterrorcode:'+JSON.stringify(ret.code));}}}exportdefault{data:{title:"",timer:null},task(){playAbility.mqttMessage(this);},mqttMessage(){this.title="开始获取MQTT消息";this.task()this.timer=setInterval(this.task,200)},stopMqtt(){clearInterval(this.timer)}}//初始化Java端Mqtt消息接收playAbility.startMqtt()Java端代码(接收Mqtt消息，异步)importorg.eclipse.paho.client.mqttv3.*;importorg.eclipse.paho.client.mqttv3.MqttMessage;importorg.eclipse.paho.client.mqttv3.persist.MemoryPersistence;importjava.util.List;publicclassMqttThreadimplementsRunnable{/**地址*/publicstaticfinalStringMQTT_BROKER_HOST="tcp://xxx.xxx.xxx.xxx:1883";/**客户端唯一标识*/publicstaticfinalStringMQTT_CLIENT_ID="client";/**预订标识*/publicstaticfinalStringMQTT_TOPIC="HarmonyTest";/**客户端*/privatevolatilestaticMqttClientmqttClient;/**连接选项*/privatestaticMqttConnectOptionsoptions;/**消息*/privatefinalListmessage;publicMqttThread(Listmessage){this.message=message;}publicvoidrun(){try{mqttClient=newMqttClient(MQTT_BROKER_HOST,MQTT_CLIENT_ID,newMemoryPersistence());options=newMqttConnectOptions();options.setCleanSession(true);options.setConnectionTimeout(20);options.setKeepAliveInterval(20);mqttClient.connect(options);mqttClient.subscribe(MQTT_TOPIC);mqttClient.setCallback(newMqttCallback(){@OverridepublicvoidconnectionLost(Throwablethrowable){}@OverridepublicvoidmessageArrived(Strings,MqttMessagemqttMessage){message.clear();message.add(mqttMessage.toString());System.out.println("接收到mqtt消息："+mqttMessage.toString());}@OverridepublicvoiddeliveryComplete(IMqttDeliveryTokeniMqttDeliveryToken){}});}catch(Exceptione){e.printStackTrace();}}}Java端代码(ParticleAbility)importcom.example.mqttapplication.mqtt.MqttThread;importohos.aafwk.ability能力；importohos.aafwk.content.Intent；importohos.hiviewdfx.HiLog；importohos.hiviewdfx.HiLogLabel；importohos.rpc.*；importohos.utils.zson.ZSONObject；importjava.util.ArrayList；importjava.util.HashMap；importjava.util.List;importjava.util.Map;publicclassPlayAbilityextendsAbility{staticfinalHiLogLabellabel=newHiLogLabel(HiLog.LOG_APP,1,"MY_TAG");privatestaticfinalintERROR=-1;privatestaticfinalintSUCCESS=0;privatestaticfinalintSTART_MQTT=1001;privatestaticfinalintMQTT_MESSAGE=1002;@OverrideprotectedvoidonStart(Intentintent){super.onStart(intent);}@OverrideprotectedIRemoteObjectonConnect(Intentintent){super.onConnect(intent);PlayRemoteremote=newPlayRemote();returnremote.asObject();}staticclassPlayRemoteextendsRemoteObjectimplementsIRemoteBroker{privateListmessage;privateThreadthread;publicPlayRemote(){super("PlayRemote");}@OverridepublicbooleanonRemoteRequest(intcode,MessageParceldata,MessageParcelreply,MessageOptionoption){//开始mqttelseif(code==START_MQTT){Mapresult=newHashMap<>();结果。put("code",SUCCESS);result.put("abilityResult","成功启动mqtt");try{message=newArrayList<>();MqttThreadmqttThread=newMqttThread(message);thread=newThread(mqttThread);线程。start();System.out.println("mqtt启动成功");}catch(Exceptione){result.put("code",ERROR);result.put("abilityResult","启动失败");}reply.writeString(ZSONObject.toZSONString(result));}//获取mqtt消息elseif(code==MQTT_MESSAGE){Mapresult=newHashMap<>();result.put("code",SUCCESS);if(message.isEmpty()){result.put("abilityResult","没有收到MQTT消息");}else{ZSONObjectzsonObject=ZSONObject.stringToZSON(message.get(0));result.put("abilityResult",zsonObject.getString("message"));}reply.writeString(ZSONObject.toZSONString(result));}else{Mapresult=newHashMap<>();result.put("abilityError",ERROR);reply.writeString(ZSONObject.toZSONString(result));returnfalse;}returntrue;}@OverridepublicIRemoteObjectasObject(){returnthis;}}}另外，要启动网络连接，需要在config.json中添加一些东西来获取权限{...”模块":{..."reqPermissions":[{"name":"ohos.permission.GET_NETWORK_INFO"},{"name":"ohos.permission.INTERNET"},{"name":"ohos.permission.SET_NETWORK_INFO"},{"名称"":"ohos.permission.MANAGE_WIFI_CONNECTION"},{"name":"ohos.permission.SET_WIFI_INFO"},{"name":"ohos.permission.GET_WIFI_INFO"}]}}最后写了一个发送mqtt消息的python脚本，很简单一行importpaho.mqtt.publishaspublishpublish.single('HarmonyTest','{"message":"BongShakalaka"}',hostname='xxx.xxx.xxx.xxx')附件：mqtt消息必须有mqttserver，这是Build自己或购买，想了解更多请访问：Harmonyos技术社区https://harmonyos.51cto.com与华为官方共建