mqtt和activemq（MQTT和ActiveMQ有什么区别）
本文目录
- MQTT和ActiveMQ有什么区别
- Flutter通过Mqtt消费ActivieMQ
- MQTT和ActiveMQ有什么区别wmqtt.jar和mqttv3.jar又有什么区别
- 各位，activemq可以限制客户端连接数吗
MQTT和ActiveMQ有什么区别
MQTT只是IBM推出的一个消息协议，基于TCP/IP的。两个App端发送和接收消息需要中间人，这个中间人就是消息服务器（比如ActiveMQ/RabbitMQ），三者通信协议就是MQTT。
wmqtt.jar是IBM实现的App端收发消息的具体实现，W意思为Webspare，说明消息服务器采用Webspare（WebSphere MQ Integrator Broker）。
Flutter通过Mqtt消费ActivieMQ
Flutter通过mqtt消费activemq，在android端主要使用插件的方式进行
处理流程
Android端连接MQTT
插件端业务处理
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 实现连接方法
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 实现mqttclient服务
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("接收到监听的消息:", "${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("连接错误:", "${e.message}")
}
})
} catch (e: MqttException) {
e.printStackTrace()
}
}
private fun subscribe() {
try {
client?.subscribe(subscribeTopic, 0) { topic, message -》
//通过广播发送监听到的消息
val intent = Intent()
intent.action = MY_ACTION
intent.putExtra("DATAPASSED", message.toString())
sendBroadcast(intent)
}
} catch (e: MqttException) {
e.printStackTrace()
}
}
}
}
step4监听广播
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")
//将监听到的消息，通过methchannel传给flutter
channel?.invokeMethod("receiveMsg", value)
} catch (e: Exception) {
e.printStackTrace()
}
}
}
}
Flutter端业务处理
实现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;
});
}
}
MQTT和ActiveMQ有什么区别wmqtt.jar和mqttv3.jar又有什么区别
ActiveMQ 这个也没有研究过，最近打算弄个聊天所以也看了下MQTT，不过也是一头雾水，看到有些人是使用 Mosquitto：An Open Source MQTT v3.1 Broker 做的（broker）代理，好像这个更简单一点儿，
还有的是用：MQTT的学习研究（二）moquette-mqtt 的使用之mqtt broker的启动
如果你只是需要推送功能的话，可以使用第三方的推送试试，比如极光，百度等，有好多的
我的理解是：MQTT他只是定义了一种通讯协议，给我们封装好了一些socket连接，让我们方便调用，像 Mosquitto 和ActiveMQ只是基于这个协议实现的代理，而wmqtt.jar 是在移动端实现接收和发送消息的接口供我们调用！
各位，activemq可以限制客户端连接数吗
可以 生产者和消费者可以使用不同的传输协议来传输消息，ActiveMQ提供了广泛的连接模式，包括HTTP/S、JGroups、JXTA、muticast、SSL、TCP、UDP、XMPP等。
《transportConnectors》 《!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB --》 《transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/》 《transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/》 《transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/》 《transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/》 《transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/》 《/transportConnectors》
可以看出最大连接数是1000，最大容量是100m
更多文章：
数据库管理系统和数据库系统分别侧重（数据库，数据库管理系统，数据库系统，这三个分别是什么意思并举个实例）
2026年9月7日 17:00
2026年9月7日 16:10
2026年9月7日 15:40
springmvc的依赖（springMVC的注入方式有哪几种，这与springMVC依赖）
2026年9月7日 14:00
2026年9月7日 13:00
2026年9月7日 12:40
display flex 自动换行（overflow-y:hidden;overflow-x:auto;无效解决方法）
2026年9月7日 11:00
2026年9月7日 10:50
timestamp without time zone（Postgresql中to_date（）函数使用问题）
2026年9月7日 09:40
2026年9月7日 04:20
