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

处理流程

Android端连接MQTT

插件端业务处理

step1:配置插件依赖包

1. implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version"
2. implementation 'com.google.code.gson:gson:2.8.5'
3. implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'
4. implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
5. implementation 'androidx.legacy:legacy-support-v4:1.0.0'
6. implementation 'androidx.appcompat:appcompat:1.1.0-rc01'
7. implementation 'androidx.constraintlayout:constraintlayout:1.1.3'
8. implementation 'androidx.recyclerview:recyclerview:1.0.0'
9. implementation 'androidx.multidex:multidex:2.0.1'
10. implementation 'androidx.multidex:multidex-instrumentation:2.0.0'
11. implementation 'com.google.android.material:material:1.1.0-alpha08'

step2 实现连接方法

1. class MqttClientPlugin : MethodCallHandler {
2. override fun onMethodCall(call: MethodCall, result: Result) {
3. when (call.method) {
4. "connectMq" -> {
5. try {
6. connectToService()
7. result.success(true)
8. } catch (e: Exception) {
9. e.printStackTrace()
10. result.success(false)
11. }
12. }
13. }
14. }
15. private fun connectToService() {
16. val intent = Intent(context, MqttClientService::class.java)
17. this.context?.startService(intent)
18. }
19. }

step3 实现mqttclient服务

1. class MqTTClientService : Service {
2. private val TAG = "ActiveMQ"
3. val clientId = "any_client_name"
4. val serverURI = "tcp://192.168.0.201:1883" //replace with your ip
5. val publishTopic = "outbox"
6. val subscribeTopic = "TJ Test"
7. var client: MqttAndroidClient? = null
8. constructor() : super()
9. val MY_ACTION = "MY_ACTION"
10. var _currentV: Int = 0
11. override fun onBind(intent: Intent?): IBinder? {
12. return null
13. }
14. override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
15. val myThread = MyThrad()
16. myThread.start()
17. return super.onStartCommand(intent, flags, startId)
18. }
19. private fun subscribe() {
20. try {
21. client?.subscribe(subscribeTopic, 0, IMqttMessageListener { topic, message ->
22. Log.i("接收到监听的消息:", "${message.payload}")
23. })
24. } catch (e: MqttException) {
25. e.printStackTrace()
26. }
27. }
28. inner class MyThrad : Thread() {
29. override fun run() {
30. val connectOptions = MqttConnectOptions()
31. connectOptions.isAutomaticReconnect = true
32. client = MqttAndroidClient(this@MqTTClientService.applicationContext, serverURI, clientId)
33. try {
34. client?.connect(connectOptions, object : IMqttActionListener {
35. override fun onSuccess(asyncActionToken: IMqttToken) {
36. subscribe()
37. }
38. override fun onFailure(asyncActionToken: IMqttToken, e: Throwable) {
39. Log.i("连接错误:", "${e.message}")
40. }
41. })
42. } catch (e: MqttException) {
43. e.printStackTrace()
44. }
45. }
46. private fun subscribe() {
47. try {
48. client?.subscribe(subscribeTopic, 0) { topic, message ->
49. //通过广播发送监听到的消息
50. val intent = Intent()
51. intent.action = MY_ACTION
52. intent.putExtra("DATAPASSED", message.toString())
53. sendBroadcast(intent)
54. }
55. } catch (e: MqttException) {
56. e.printStackTrace()
57. }
58. }
59. }
60. }

step4监听广播

1. class DevicemanagerPlugin : MethodCallHandler {
2. constructor(context: Context?, channel: MethodChannel) {
3. this.context = context
4. this.channel = channel
5. initMqtt()
6. register()
7. }
8. private fun register() {
9. myReceiver = MyReceiver()
10. val intentFilter = IntentFilter()
11. intentFilter.addAction("MY_ACTION")
12. this.context?.registerReceiver(myReceiver, intentFilter)
13. }
14. inner class MyReceiver : BroadcastReceiver() {
15. override fun onReceive(context: Context?, intent: Intent?) {
16. try {
17. val value = intent?.getStringExtra("DATAPASSED")
18. //将监听到的消息，通过methchannel传给flutter
19. channel?.invokeMethod("receiveMsg", value)
20. } catch (e: Exception) {
21. e.printStackTrace()
22. }
23. }
24. }
25. }

Flutter端业务处理

实现receiveMsg方法

1. const channel = const MethodChannel("mqttclient");
2. class _MyHomePageState extends State<MyHomePage> {
3. @override
4. void initState() {
5. registerMethod()
6. connectActiveMq()
7. }
8. void connectActiveMq() async{
9. if (Platform.isAndroid) {
10. var result = await Devicemanager.connectMq(API.MQ_URI);
11. if (result) {
12. print("mq链接成功");
13. } else {
14. print("mq链接失败");
15. }
16. }
17. }
18. void registerMethod() {
19. channel.setMethodCallHandler((handler) {
20. var completer = new Completer<String>();
21. try {
22. switch (handler.method) {
23. case "receiveMsg":
24. //接收到的消息
25. var v = handler.arguments;
26. break;
27. default:
28. break;
29. }
30. } catch (e) {
31. print(e);
32. }
33. return completer.future;
34. });
35. }
36. }