使用一個(gè)開(kāi)源庫(kù)：hivemq-mqtt-client，這是Java生態(tài)的一個(gè)MQTT客戶端框架，需要Java 8，Android上使用的話問(wèn)題不大，需要一些額外的配置，下面列出了相關(guān)的配置，尤其是 packagingOptions，不然編譯不過(guò)，因?yàn)榭蚣苁褂昧薐ava8新增的語(yǔ)言特性，所以 minSdk 設(shè)置為24，即Android7.0，如果要兼容Android7.0以下系統(tǒng)，可以參考這份詳細(xì)文檔配置一下語(yǔ)法脫糖的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'
}

剛開(kāi)始在自動(dòng)連接這塊花了好多時(shí)間，最后才發(fā)現(xiàn)是設(shè)置用戶名和密碼的地方不對(duì)，一定要在設(shè)置自動(dòng)重連（初始化Client）的地方設(shè)置，而不是連接的時(shí)候！下面是一個(gè)簡(jiǎn)單的使用示例代碼

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()//在初始化的時(shí)候設(shè)置賬號(hào)密碼，重連才能成功.username(USERNAME).password(PASSWORD.toByteArray()).applySimpleAuth().automaticReconnectWithDefaultConfig()//自動(dòng)重連.buildAsync()private val listeners = mutableListOf<MqttListener>()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()//等待所有訂閱結(jié)果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 = Consumer<Mqtt5Publish> {val topic = it.topic.toString()val data = it.payloadAsBytesprocessReceivedMessage(topic, data)}private fun processReceivedMessage(topic: String, data: ByteArray) {//處理接收的數(shù)據(jù)for (l in listeners) {l.onReceiveMessage(topic, data)}}fun subscribe(topic: String): CompletableFuture<Mqtt5SubAck> {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")}}/*** 發(fā)送數(shù)據(jù)*/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 = "MqttManager"private const val SERVER_HOST = "example.com"private const val SERVER_PORT = 1883 // 1883即TCP協(xié)議，host不要再加上"tcp://"，否則連不成功private const val USERNAME = "admin"private const val PASSWORD = "123456"val instance = MqttManager()}
}