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 部分。
1dependencies { 2 implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.4' 3 implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1' 4}
配置 AndroidManifest.xml
Android Service 是 Eclipse 开发的基于 Android 平台的一个后台服务，我们需要将它注册到AndroidManifest.xml 文件，同时，我们需要注册权限。
1<uses-permission android:name="android.permission.INTERNET" /> 2<uses-permission android:name="android.permission.WAKE_LOCK" /> 3<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" /> 4 5<application 6 ... 7 <service android:name="org.eclipse.paho.android.service.MqttService" /> 8</application> 9
创建 MQTT 客户端
1private lateinit var mqttClient: MqttAndroidClient 2// TAG 3companion object { 4 const val TAG = "AndroidMqttClient" 5}
连接 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") }
1 override fun connectionLost(cause: Throwable?) { 2 Log.d(TAG, "Connection lost ${cause.toString()}") 3 } 4 5 override fun deliveryComplete(token: IMqttDeliveryToken?) { 6 7 } 8 }) 9 val options = MqttConnectOptions() 10 try { 11 mqttClient.connect(options, null, object : IMqttActionListener { 12 override fun onSuccess(asyncActionToken: IMqttToken?) { 13 Log.d(TAG, "Connection success") 14 } 15 16 override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) { 17 Log.d(TAG, "Connection failure") 18 } 19 }) 20 } catch (e: MqttException) { 21 e.printStackTrace() 22 } 23 24}
其中，MqttCallback 接口包含 3 个方法：
- messageArrived：收到 broker 新消息
- connectionLost：与 broker 连接丢失
- deliveryComplete：消息到 broker 传递完成
MqttConnectOptions 用于配置连接设置，包含用户名密码，超时配置等，具体可以查看其方法。
创建 MQTT 订阅
订阅 topic
1fun subscribe(topic: String, qos: Int = 1) { 2 try { 3 mqttClient.subscribe(topic, qos, null, object : IMqttActionListener { 4 override fun onSuccess(asyncActionToken: IMqttToken?) { 5 Log.d(TAG, "Subscribed to $topic") 6 } 7 8 override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) { 9 Log.d(TAG, "Failed to subscribe $topic") 10 } 11 }) 12 } catch (e: MqttException) { 13 e.printStackTrace() 14 } 15 }
取消订阅
取消订阅 topic
1fun unsubscribe(topic: String) { 2 try { 3 mqttClient.unsubscribe(topic, null, object : IMqttActionListener { 4 override fun onSuccess(asyncActionToken: IMqttToken?) { 5 Log.d(TAG, "Unsubscribed to $topic") 6 } 7 8 override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) { 9 Log.d(TAG, "Failed to unsubscribe $topic") 10 } 11 }) 12 } catch (e: MqttException) { 13 e.printStackTrace() 14 } 15 }
发布消息
1fun publish(topic: String, msg: String, qos: Int = 1, retained: Boolean = false) { 2 try { 3 val message = MqttMessage() 4 message.payload = msg.toByteArray() 5 message.qos = qos 6 message.isRetained = retained 7 mqttClient.publish(topic, message, null, object : IMqttActionListener { 8 override fun onSuccess(asyncActionToken: IMqttToken?) { 9 Log.d(TAG, "$msg published to $topic") 10 } 11 12 override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) { 13 Log.d(TAG, "Failed to publish $msg to $topic") 14 } 15 }) 16 } catch (e: MqttException) { 17 e.printStackTrace() 18 } 19 }
断开 MQTT 连接
1fun disconnect() { 2 try { 3 mqttClient.disconnect(null, object : IMqttActionListener { 4 override fun onSuccess(asyncActionToken: IMqttToken?) { 5 Log.d(TAG, "Disconnected") 6 } 7 8 override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) { 9 Log.d(TAG, "Failed to disconnect") 10 } 11 }) 12 } catch (e: MqttException) { 13 e.printStackTrace() 14 } 15 }
测试
首先将 Android 客户端连接到 MQTT 服务器，然后订阅 topic: a/b，可以看到连接成功和成功订阅的日志
然后我们使用 MQTT 5.0 客户端工具 - MQTT X 进行测试，发布消息到 topic: a/b，客户端可以看到收到消息的日志
我们在客户端发布消息到 topic: a/b ，因为我们订阅了该 topic，同时也会收到消息，最后我们断开客户端与 MQTT 服务器的连接，日志如下
至此，我们已经完成了Android 上 MQTT 客户端的构建，实现了客户端与 MQTT 服务器的连接、主题订阅、收发消息等功能。
MQTT 可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。
而 Kotlin 也是 Google 官方主推的一门语言，结合 MQTT 协议及 MQTT 云服务，我们可以开发更多有趣的应用。