Android 使用 Kotlin 连接 MQTT - HelloWorld开发者社区

MQTT 是一种轻量级的、灵活的物联网消息交换和数据传递协议，致力于为 IoT 开发人员实现灵活性与硬件/网络资源的平衡。

Kotlin 是一门由 JetBrains 公司开发的编程语言，Kotlin 是基于 JVM 的，所以开发者可以很方便地用它来进行Android 开发，并且支持 Kotlin 和 Java 的混合编写。而早在 2017 年，Google 就宣布 Kotlin 成为官方开发语言。

本文主要介绍使用 Kotlin 语言在 Android 平台上使用 MQTT。

新建 Kotlin 项目

打开 Android Studio 新建一个项目，选择语言为 Kotlin，Android Studio 会自动创建 Kotlin 相关配置。若要配置现有项目，则可以参考将 Kotlin 添加到现有应用。

添加依赖

打开项目的 build.gradle，添加 Eclipse Paho Java Client 和 Eclipse Paho Android Service 依赖到 dependencies 部分。

dependencies {
    implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.4'
    implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1' 
}

配置 `AndroidManifest.xml`

Android Service 是 Eclipse 开发的基于 Android 平台的一个后台服务，我们需要将它注册到AndroidManifest.xml 文件，同时，我们需要注册权限。

<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.WAKE_LOCK" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />

<application
   ...
   <service android:name="org.eclipse.paho.android.service.MqttService" />
</application>

创建 MQTT 客户端

private lateinit var mqttClient: MqttAndroidClient
// TAG
companion object {
    const val TAG = "AndroidMqttClient"
}

连接 MQTT 服务器

本文将使用 EMQ X MQTT Cloud 运营和维护的免费公共 MQTT 服务器， EMQ X Cloud 是由 EMQ 推出的安全的 MQTT 物联网云服务平台，它提供一站式运维代管、独有隔离环境的 MQTT 5.0 接入服务。

Broker: broker.emqx.io
TCP Port: 1883
Websocket Port: 8083

fun connect(context: Context) { val serverURI = "tcp://broker.emqx.io:1883" mqttClient = MqttAndroidClient(context, serverURI, "kotlin_client") mqttClient.setCallback(object : MqttCallback { override fun messageArrived(topic: String?, message: MqttMessage?) { Log.d(TAG, "Receive message: ${message.toString()} from topic: $topic") } override fun connectionLost(cause: Throwable?) { Log.d(TAG, "Connection lost ${cause.toString()}") } override fun deliveryComplete(token: IMqttDeliveryToken?) { } }) val options = MqttConnectOptions() try { mqttClient.connect(options, null, object : IMqttActionListener { override fun onSuccess(asyncActionToken: IMqttToken?) { Log.d(TAG, "Connection success") } override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) { Log.d(TAG, "Connection failure") } }) } catch (e: MqttException) { e.printStackTrace() } }

其中，MqttCallback 接口包含 3 个方法：

messageArrived：收到 broker 新消息
connectionLost：与 broker 连接丢失
deliveryComplete：消息到 broker 传递完成

MqttConnectOptions 用于配置连接设置，包含用户名密码，超时配置等，具体可以查看其方法。

创建 MQTT 订阅

订阅 topic

fun subscribe(topic: String, qos: Int = 1) {
        try {
            mqttClient.subscribe(topic, qos, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.d(TAG, "Subscribed to $topic")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.d(TAG, "Failed to subscribe $topic")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

取消订阅

取消订阅 topic

fun unsubscribe(topic: String) {
        try {
            mqttClient.unsubscribe(topic, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.d(TAG, "Unsubscribed to $topic")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.d(TAG, "Failed to unsubscribe $topic")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

发布消息

fun publish(topic: String, msg: String, qos: Int = 1, retained: Boolean = false) {
        try {
            val message = MqttMessage()
            message.payload = msg.toByteArray()
            message.qos = qos
            message.isRetained = retained
            mqttClient.publish(topic, message, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.d(TAG, "$msg published to $topic")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.d(TAG, "Failed to publish $msg to $topic")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

断开 MQTT 连接

fun disconnect() {
        try {
            mqttClient.disconnect(null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.d(TAG, "Disconnected")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.d(TAG, "Failed to disconnect")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }