网站设计图尺寸,网站建设教程资源,网址大全黄页男女,网络舆情监测存在的问题使用一个开源库#xff1a;hivemq-mqtt-client#xff0c;这是Java生态的一个MQTT客户端框架#xff0c;需要Java 8#xff0c;Android上使用的话问题不大#xff0c;需要一些额外的配置#xff0c;下面列出了相关的配置#xff0c;尤其是 packagingOptions#xff0c;…使用一个开源库hivemq-mqtt-client这是Java生态的一个MQTT客户端框架需要Java 8Android上使用的话问题不大需要一些额外的配置下面列出了相关的配置尤其是 packagingOptions不然编译不过因为框架使用了Java8新增的语言特性所以 minSdk 设置为24即Android7.0如果要兼容Android7.0以下系统可以参考这份详细文档配置一下语法脱糖的SDK Installation on Android
android {defaultConfig {minSdk 24}compileOptions {sourceCompatibility JavaVersion.VERSION_8targetCompatibility JavaVersion.VERSION_8}kotlinOptions {jvmTarget 8}packagingOptions {resources {excludes [META-INF/INDEX.LIST, META-INF/io.netty.versions.properties]}}
}dependencies {implementation com.hivemq:hivemq-mqtt-client:1.3.3
}
刚开始在自动连接这块花了好多时间最后才发现是设置用户名和密码的地方不对一定要在设置自动重连初始化Client的地方设置而不是连接的时候下面是一个简单的使用示例代码
MqttManager.kt
import android.util.Log
import com.hivemq.client.mqtt.datatypes.MqttQos
import com.hivemq.client.mqtt.lifecycle.MqttClientConnectedContext
import com.hivemq.client.mqtt.lifecycle.MqttClientConnectedListener
import com.hivemq.client.mqtt.lifecycle.MqttClientDisconnectedContext
import com.hivemq.client.mqtt.lifecycle.MqttClientDisconnectedListener
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAckReasonCode
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck
import java.util.UUID
import java.util.concurrent.CompletableFuture
import java.util.concurrent.Executors
import java.util.function.Consumeropen class MqttListener {open fun onConnected() {}open fun onDisconnected() {}open fun onSubscribed(vararg topics: String) {}open fun onReceiveMessage(topic: String, data: ByteArray) {}open fun onSendMessage(topic: String, data: ByteArray) {}
}/*
文档
https://github.com/hivemq/hivemq-mqtt-client
https://hivemq.github.io/hivemq-mqtt-client/docs/installation/android/
*/
class MqttManager private constructor() : MqttClientConnectedListener, MqttClientDisconnectedListener {private val executor Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()) {Thread(it).apply { isDaemon true }}private val mqttAsynClient: Mqtt5AsyncClient Mqtt5Client.builder().identifier(UUID.randomUUID().toString()).serverHost(SERVER_HOST).serverPort(SERVER_PORT).addConnectedListener(this).addDisconnectedListener(this).simpleAuth()//在初始化的时候设置账号密码重连才能成功.username(USERNAME).password(PASSWORD.toByteArray()).applySimpleAuth().automaticReconnectWithDefaultConfig()//自动重连.buildAsync()private val listeners mutableListOfMqttListener()private val subTopicsget() arrayOf(top1, top2, top3)fun addMqttListener(listener: MqttListener) {if (!listeners.contains(listener)) {listeners.add(listener)}}fun removeMqttListener(listener: MqttListener) {listeners.remove(listener)}override fun onConnected(context: MqttClientConnectedContext) {Log.i(TAG, onConnected())for (l in listeners) {l.onConnected()}subscribeAll()}private fun subscribeAll() {CompletableFuture.supplyAsync({val futures subTopics.map(::subscribe).map {it.thenCompose {CompletableFuture.supplyAsync({val success !it.reasonString.isPresentif (success) {Log.i(TAG, subscribe success)} else {Log.e(TAG, subscribe() - reasonCodes[${it.reasonCodes.joinToString(, )}] , reasonString${it.reasonString})}success}, executor)}}.toTypedArray()CompletableFuture.allOf(*futures).join()//等待所有订阅结果if(futures.all { it.get() }) {Log.i(TAG, subscribeAll() - 全部订阅成功)}for (l in listeners) {l.onSubscribed(*subTopics)}}, executor)}override fun onDisconnected(context: MqttClientDisconnectedContext) {Log.e(TAG, onDisconnected() - isConnected${mqttAsynClient.state.isConnected} , isConnectedOrReconnect${mqttAsynClient.state.isConnectedOrReconnect})for (l in listeners) {l.onDisconnected()}}fun connect() {mqttAsynClient.connectWith().cleanStart(true).keepAlive(30).send().thenAccept {if (it.reasonCode Mqtt5ConnAckReasonCode.SUCCESS) {Log.i(TAG, connect() - SUCCESS)} else {Log.e(TAG, connect() - ${it.reasonCode})}}}fun disconnect() {mqttAsynClient.disconnect().thenAccept {Log.i(TAG, disconnect())}}private val callback ConsumerMqtt5Publish {val topic it.topic.toString()val data it.payloadAsBytesprocessReceivedMessage(topic, data)}private fun processReceivedMessage(topic: String, data: ByteArray) {//处理接收的数据for (l in listeners) {l.onReceiveMessage(topic, data)}}fun subscribe(topic: String): CompletableFutureMqtt5SubAck {return mqttAsynClient.subscribeWith().topicFilter(topic).noLocal(true)// we do not want to receive our own message.qos(MqttQos.AT_MOST_ONCE).callback(callback).executor(executor).send()}fun unsubscribe(topic: String) {mqttAsynClient.unsubscribeWith().topicFilter(topic).send().thenAccept {Log.i(TAG, unsubscribe() - $it)}}/*** 发送数据*/fun publish(topic: String, payload: ByteArray) {mqttAsynClient.publishWith().topic(topic).qos(MqttQos.AT_MOST_ONCE).payload(payload).send().thenAccept { mqtt5PublishResult -mqtt5PublishResult.publish.let { mqtt5Publish -
// val topic mqtt5Publish.topic.toString()val data mqtt5Publish.payloadAsBytesfor (l in listeners) {l.onSendMessage(topic, data)}}}}companion object {private const val TAG MqttManagerprivate const val SERVER_HOST example.comprivate const val SERVER_PORT 1883 // 1883即TCP协议host不要再加上tcp://否则连不成功private const val USERNAME adminprivate const val PASSWORD 123456val instance MqttManager()}
}
