Android MQTT Client

2024-04-08

在上篇《ActiveMQ MQTT Server 搭建》中我们已经搭建好了 MQTT Server，本篇我们将在 Android 上建立 MQTT Client 与 MQTT Server 进行通讯。

1. 执行流程

1.1 引入依赖

在 build.gradle 中引入依赖项。

 android {
     ...
 }
 
 repositories {
    maven {
        url "https://repo.eclipse.org/content/repositories/paho-snapshots/"
    }
}

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

1.2 配置清单

在 AndroidManifest.xml 中声明权限并引入 MqttService 配置。

<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE"/>
<uses-permission android:name="android.permission.INTERNET"/>
<uses-permission android:name="android.permission.WAKE_LOCK"/>
    
<application>
    <service android:name="org.eclipse.paho.android.service.MqttService"/>
</application>

1.3 增加回调

class MqttCallback : MqttCallbackExtended {

    override fun connectComplete(reconnect: Boolean, serverURI: String?) {
        Log.i(TAG, "connect Complete " + Thread.currentThread().id)
    }

    override fun messageArrived(topic: String?, message: MqttMessage?) {
        Log.i(TAG, "topic: " + topic + " message: " + String(message!!.payload, Charset.forName("utf-8")))

    }

    override fun connectionLost(cause: Throwable?) {
        Log.i(TAG, "connection Lost")
    }

    override fun deliveryComplete(token: IMqttDeliveryToken?) {
        Log.i(TAG, "delivery Complete")
    }
}

✦ 注意： 回调的四个方法中，不能进行 UI 的更新操作，可使用 Handler 进行 UI 的更新操作。

1.4 初始化 MQTT Client

在《ActiveMQ MQTT Server 搭建》中，服务启动成功后会在控制台输出 MQTT 的监听信息，该信息包含主机名及端口号，如下图所示，MQTT Server 的地址应设置为：tcp://xxx.xxx.xxx.xxx:1883，其中 1883 为端口号，前面为对应服务器的 IP。

此处输入图片的描述

private var mMqttClient: MqttAsyncClient? = null
private val host = "tcp://192.168.0.98:1883"
private val mClientID = "xx:xx:xx:xx:xx:xx"
...

if (null == mMqttClient) {
    mMqttClient = MQTTUtils.initClient(host, mClientID, MqttCallback())
}

我们使用一个公用 MQTTHelper 来处理初始化。

object MQTTHelper {
    fun initClient(serverURI: String, clientId: String, callback: MqttCallback): MqttAsyncClient {
        return MqttAsyncClient(serverURI, clientId, MemoryPersistence()).apply {
            setCallback(callback)
        }
    }
}

参数	说明
serverURI	MQTT Server 地址
clientId	唯一识别码，告知 Broker 知道是哪个 Client
callback	回调函数

1.5 设置连接参数

object MQTTUtils {
    fun initMqttConnectionOptions(): MqttConnectOptions {
        return MqttConnectOptions().apply {
            isAutomaticReconnect = false
            isCleanSession = true
            connectionTimeout = 10
            userName = "admin"
            password = "admin".toCharArray()
            keepAliveInterval = 10
            maxInflight = 10
            mqttVersion = MqttConnectOptions.MQTT_VERSION_3_1_1
        }
    }
}

参数	说明
isAutomaticReconnect	连接中断时，是否会自动重连
isCleanSession	断开连接时是否清除连接信息
connectionTimeout	连接超时时间，以秒为单位
userName	账号
password	密码
keepAliveInterval	心跳间隔，每隔几秒会再度确认是否连接
maxInflight	允許同時发送几条信息
mqttVersion	MQTT 版本

1.6 连接 MQTT Server

private val mTopicID = "topic1"

val options = MQTTHelper.initMqttConnectionOptions()

mMqttClient?.let {
    it.connect(options, null, object : IMqttActionListener {
        override fun onSuccess(asyncActionToken: IMqttToken) {
            it.subscribe(mTopicID, 1)
        }

        override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {
        }
    })
}

✦ 说明： 连接成功后，会向服务端发送 Topic。

2. 执行结果

连接成功后，通过 ActiveMQ 的管理界面，可以看到 Topics 有刚刚设定的 topic1。

此处输入图片的描述

点击 topic1 即可进入消息发送页面，向 Android 设备发送消息。

此处输入图片的描述

3. 全部源码

class MainActivity : AppCompatActivity() {

    private var mMqttClient: MqttAsyncClient? = null
    private val host = "tcp://192.168.61.32:1883"
    private val mClientID = "xx:xx:xx:xx:xx:xx"
    private val mTopicID = "topic1"

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        findViewById<TextView>(R.id.connect).setOnClickListener {
            connect()
        }
    }

    override fun onDestroy() {
        super.onDestroy()
        mMqttClient?.disconnect()
    }

    private fun connect() {
        Thread {
            try {
                if (mMqttClient == null) {
                    mMqttClient = MQTTHelper.initClient(host, mClientID, MqttCallback())
                }
                val options = MQTTHelper.initMqttConnectionOptions()

                mMqttClient?.let {
                    it.connect(options, null, object : IMqttActionListener {
                        override fun onSuccess(asyncActionToken: IMqttToken) {
                            it.subscribe(mTopicID, 1)
                        }

                        override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {
                        }
                    })
                }
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }.start()
    }

    class MqttCallback : MqttCallbackExtended {

        private val TAG = "MqttCallback"

        override fun connectComplete(reconnect: Boolean, serverURI: String?) {
            Log.i(TAG, "connect Complete " + Thread.currentThread().id)
        }

        override fun messageArrived(topic: String?, message: MqttMessage?) {
            Log.i(TAG, "topic: " + topic + " message: " + String(message!!.payload, Charset.forName("utf-8")))

        }

        override fun connectionLost(cause: Throwable?) {
            Log.i(TAG, "connection Lost ")
        }

        override fun deliveryComplete(token: IMqttDeliveryToken?) {
            Log.i(TAG, "delivery Complete ")
        }
    }

}

object MQTTHelper {

    fun initClient(serverURI: String, clientId: String, callback: MainActivity.MqttCallback): MqttAsyncClient {
        return MqttAsyncClient(serverURI, clientId, MemoryPersistence()).apply {
            setCallback(callback)
        }
    }

    fun initMqttConnectionOptions(): MqttConnectOptions {
        return MqttConnectOptions().apply {
            isAutomaticReconnect = false
            isCleanSession = true
            connectionTimeout = 10
            userName = "admin"
            password = "admin".toCharArray()
            keepAliveInterval = 10
            maxInflight = 10
            mqttVersion = MqttConnectOptions.MQTT_VERSION_3_1_1
        }
    }

}