相比于 mqtt-client,mqttv3 使用的人相对多些,如果出现问题,好排查一些。
activemq 部署 MQTT 服务
查看文件:conf\activemq.xml,
如果包含下面内容,activemq 本身已经包含 MQTT 服务,不需要任何其它配置。
activemq 不局限于下面这些,还可以继续扩展,比如:NIO、SSL。
前往官网查看:https://activemq.apache.org/components/classic/documentation/auto
<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>
Maven 依赖
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
关键参数
CleanSession 清除会话
设置为 false 之后,在 client 断线恢复之后,能接收到期间错过的消息。
注意磁盘空间和会话保持时间,避免磁盘空间不足,或者超时自动清除。
Qos 服务质量
- QoS 0:最多一次传输,消息可能会丢失。
- QoS 1:至少一次传输,消息可能会重复。
- QoS 2:仅一次传输,确保消息不会重复也不会丢失。
消息保留 retain
服务器会保留最新的一条 retain 值为 true 的消息(只有最后一条);
这个机制,可以确保当客户端连接到 MQTT 服务器时,可以立即获取队列最新状态。
代码样例
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
/**
* 发送 MQTT 消息
*
* @author Mr.css
* @version 2024-10-15 9:16
*/
public class MqttPublish {
public static void main(String[] args) throws Exception {
try {
String topic = "mqtt";
String broker = "tcp://localhost:1883";
String clientId = "MqttPublish";
MqttClient client = new MqttClient(broker, clientId);
// 使用自定义参数连接
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("dev");
options.setPassword("dev".toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
// 是否清理会话,这个配置非常重要,置为 false,client 可以获取到断线期间的消息
options.setCleanSession(false);
client.connect(options);
String content = "Hello MQTT";
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(2);
message.setRetained(false);
client.publish(topic, message);
System.out.println("send: " + message);
client.disconnect();
} catch (MqttException e) {
e.printStackTrace();
}
}
}
import org.eclipse.paho.client.mqttv3.*;
/**
* 接收 MQTT 消息
*
* @author Mr.css
* @version 2024-10-15 9:16
*/
public class MQTTReceiver implements MqttCallback {
public static void main(String[] args) {
try {
String topic = "mqtt";
String broker = "tcp://localhost:1883";
String clientId = "MQTTReceiver";
MqttClient client = new MqttClient(broker, clientId);
MqttCallback callback = new MQTTReceiver();
client.setCallback(callback);
// 使用自定义参数连接
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("dev");
options.setPassword("dev".toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
client.connect(options);
int qos = 2;
client.subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
@Override
public void connectionLost(Throwable cause) {
// 当连接丢失时调用
System.out.println("Connection lost");
cause.printStackTrace();
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
// 当接收到消息时调用
System.out.println("Message arrived: " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
// 当消息发送成功时调用
System.out.println("Delivery complete");
}
}
标签:String,mqttv3,client,org,message,activemq,options
From: https://www.cnblogs.com/chenss15060100790/p/18524419