Flutter通過(guò)mqtt消費(fèi)activemq，在android端主要使用插件的方式進(jìn)行

處理流程

Android端連接MQTT

插件端業(yè)務(wù)處理

step1:配置插件依賴包

implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version"

implementation 'com.google.code.gson:gson:2.8.5'

implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'

implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

implementation 'androidx.legacy:legacy-support-v4:1.0.0'

implementation 'androidx.appcompat:appcompat:1.1.0-rc01'

implementation 'androidx.constraintlayout:constraintlayout:1.1.3'

implementation 'androidx.recyclerview:recyclerview:1.0.0'

implementation 'androidx.multidex:multidex:2.0.1'

implementation 'androidx.multidex:multidex-instrumentation:2.0.0'

implementation 'com.google.android.material:material:1.1.0-alpha08'

step2 實(shí)現(xiàn)連接方法

class MqttClientPlugin : MethodCallHandler {

override fun onMethodCall(call: MethodCall, result: Result) {

when (call.method) {

"connectMq" -> {

try {

connectToService()

result.success(true)

} catch (e: Exception) {

e.printStackTrace()

result.success(false)

}

}

}

}

private fun connectToService() {

val intent = Intent(context, MqttClientService::class.java)

this.context?.startService(intent)

}

}

step3 實(shí)現(xiàn)mqttclient服務(wù)

class MqTTClientService : Service {

private val TAG = "ActiveMQ"

val clientId = "any_client_name"

val serverURI = "tcp://192.168.0.201:1883" //replace with your ip

val publishTopic = "outbox"

val subscribeTopic = "TJ Test"

var client: MqttAndroidClient? = null

constructor() : super()

val MY_ACTION = "MY_ACTION"

var _currentV: Int = 0

override fun onBind(intent: Intent?): IBinder? {

return null

}

override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {

val myThread = MyThrad()

myThread.start()

return super.onStartCommand(intent, flags, startId)

}

private fun subscribe() {

try {

client?.subscribe(subscribeTopic, 0, IMqttMessageListener { topic, message ->

Log.i("接收到監(jiān)聽(tīng)的消息:", "${message.payload}")

})

} catch (e: MqttException) {

e.printStackTrace()

}

}

inner class MyThrad : Thread() {

override fun run() {

val connectOptions = MqttConnectOptions()

connectOptions.isAutomaticReconnect = true

client = MqttAndroidClient(this@MqTTClientService.applicationContext, serverURI, clientId)

try {

client?.connect(connectOptions, object : IMqttActionListener {

override fun onSuccess(asyncActionToken: IMqttToken) {

subscribe()

}

override fun onFailure(asyncActionToken: IMqttToken, e: Throwable) {

Log.i("連接錯(cuò)誤:", "${e.message}")

}

})

} catch (e: MqttException) {

e.printStackTrace()

}

}

private fun subscribe() {

try {

client?.subscribe(subscribeTopic, 0) { topic, message ->

//通過(guò)廣播發(fā)送監(jiān)聽(tīng)到的消息

val intent = Intent()

intent.action = MY_ACTION

intent.putExtra("DATAPASSED", message.toString())

sendBroadcast(intent)

}

} catch (e: MqttException) {

e.printStackTrace()

}

}

}

}

step4監(jiān)聽(tīng)廣播

class DevicemanagerPlugin : MethodCallHandler {

constructor(context: Context?, channel: MethodChannel) {

this.context = context

this.channel = channel

initMqtt()

register()

}

private fun register() {

myReceiver = MyReceiver()

val intentFilter = IntentFilter()

intentFilter.addAction("MY_ACTION")

this.context?.registerReceiver(myReceiver, intentFilter)

}

inner class MyReceiver : BroadcastReceiver() {

override fun onReceive(context: Context?, intent: Intent?) {

try {

val value = intent?.getStringExtra("DATAPASSED")

//將監(jiān)聽(tīng)到的消息，通過(guò)methchannel傳給flutter

channel?.invokeMethod("receiveMsg", value)

} catch (e: Exception) {

e.printStackTrace()

}

}

}

}

Flutter端業(yè)務(wù)處理

實(shí)現(xiàn)receiveMsg方法

const channel = const MethodChannel("mqttclient");

class _MyHomePageState extends State<MyHomePage> {

@override

void initState() {

registerMethod()

connectActiveMq()

}

void connectActiveMq() async{

if (Platform.isAndroid) {

var result = await Devicemanager.connectMq(API.MQ_URI);

if (result) {

print("mq鏈接成功");

} else {

print("mq鏈接失敗");

}

}

}

void registerMethod() {

channel.setMethodCallHandler((handler) {

var completer = new Completer<String>();

try {

switch (handler.method) {

case "receiveMsg":

//接收到的消息

var v = handler.arguments;

break;

default:

break;

}

} catch (e) {

print(e);

}

return completer.future;

});

}

}

總結(jié)

以上是生活随笔為你收集整理的mqtt连接失败_Flutter通过Mqtt消费ActivieMQ的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。