Flutter通过Mqtt消费ActivieMQ

Flutter通过mqtt消费activemq，在android端主要使用插件的方式进行

处理流程

Android端连接MQTT

插件端业务处理

step1:配置插件依赖包

    implementation "orgjetbrainskotlin:kotlin-stdlib-jdk7:$kotlin_version"

    implementation 'comgooglecodegson:gson:285'

    implementation 'orgeclipsepaho:orgeclipsepahoclientmqttv3:110'

    implementation 'orgeclipsepaho:orgeclipsepahoandroidservice:111'

    implementation 'androidxlegacy:legacy-support-v4:100'

    implementation 'androidxappcompat:appcompat:110-rc01'

    implementation 'androidxconstraintlayout:constraintlayout:113'

    implementation 'androidxrecyclerview:recyclerview:100'

    implementation 'androidxmultidex:multidex:201'

    implementation 'androidxmultidex:multidex-instrumentation:200'

    implementation 'comgoogleandroidmaterial:material:110-alpha08'

step2 实现连接方法

class MqttClientPlugin : MethodCallHandler {

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

        when (callmethod) {

            "connectMq" -> {

                    try {

                        connectToService()

                        resultsuccess(true)

                    } catch (e: Exception) {

                        eprintStackTrace()

                        resultsuccess(false)

                    }

                }

        }

    }

    private fun connectToService() {

        val intent = Intent(context, MqttClientService::classjava)

        thiscontextstartService(intent)

    }

}

step3 实现mqttclient服务

class MqTTClientService : Service {

    private val TAG = "ActiveMQ"

    val clientId = "any_client_name"

    val serverURI = "tcp://1921680201: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()

        myThreadstart()

        return superonStartCommand(intent, flags, startId)

    }

    private fun subscribe() {

        try {

            clientsubscribe(subscribeTopic, 0, IMqttMessageListener { topic, message ->

                Logi("接收到监听的消息:", "${messagepayload}")

            })

        } catch (e: MqttException) {

            eprintStackTrace()

        }

    }

    inner class MyThrad : Thread() {

        override fun run() {

            val connectOptions = MqttConnectOptions()

            connectOptionsisAutomaticReconnect = true

            client = MqttAndroidClient(this@MqTTClientServiceapplicationContext, serverURI, clientId)

            try {

                clientconnect(connectOptions, object : IMqttActionListener {

                    override fun onSuccess(asyncActionToken: IMqttToken) {

                        subscribe()

                    }

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

                        Logi("连接错误:", "${emessage}")

                    }

                })

            } catch (e: MqttException) {

                eprintStackTrace()

            }

        }

        private fun subscribe() {

            try {

                clientsubscribe(subscribeTopic, 0) { topic, message ->

                    //通过广播发送监听到的消息

                    val intent = Intent()

                    intentaction = MY_ACTION

                    intentputExtra("DATAPASSED", messagetoString())

                    sendBroadcast(intent)

                }

            } catch (e: MqttException) {

                eprintStackTrace()

            }

        }

    }

}

step4监听广播

class DevicemanagerPlugin : MethodCallHandler {

    constructor(context: Context, channel: MethodChannel) {

            thiscontext = context

            thischannel = channel

            initMqtt()

            register()

        }

    private fun register() {

            myReceiver = MyReceiver()

            val intentFilter = IntentFilter()

            intentFilteraddAction("MY_ACTION")

            thiscontextregisterReceiver(myReceiver, intentFilter)

        }

    inner class MyReceiver : BroadcastReceiver() {

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

            try {

                val value = intentgetStringExtra("DATAPASSED")

                //将监听到的消息，通过methchannel传给flutter

                channelinvokeMethod("receiveMsg", value)

            } catch (e: Exception) {

                eprintStackTrace()

            }

        }

    }   

}

Flutter端业务处理

实现receiveMsg方法

const channel = const MethodChannel("mqttclient");

class _MyHomePageState extends State<MyHomePage> {

    @override

  void initState() {

      registerMethod()

      connectActiveMq()

  }

  void connectActiveMq() async{

    if (PlatformisAndroid) {

      var result = await DevicemanagerconnectMq(APIMQ_URI);

      if (result) {

        print("mq链接成功");

      } else {

        print("mq链接失败");

      }

    }

  }

  void registerMethod() {

    channelsetMethodCallHandler((handler) {

      var completer = new Completer<String>();

      try {

        switch (handlermethod) {

          case "receiveMsg":

            //接收到的消息

            var v = handlerarguments;

            break;

          default:

            break;

        }

      } catch (e) {

        print(e);

      }

      return completerfuture;

    });

  }

}

varsocketOpen=falsevarsocketMsgQueue=[]functionsendSocketMessage(msg){consolelog('sendmsg:')consolelog(msg);if(socketOpen){wxsendSocketMessage({data:msg

以上就是关于Flutter通过Mqtt消费ActivieMQ全部的内容，包括:Flutter通过Mqtt消费ActivieMQ、如何在微信小程序的websocket上使用mqtt协议、等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！