yangle-wangannan
/
mqtt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat 发布 MQTT 消息 和订阅 MQTT 主题

master
rouchen 2024-05-31 22:21:40 +08:00
parent fceab89ed4
commit fa25c11a79
4 changed files with 162 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/muyu/mqtt/MsgHandler.java
View File

@ -23,7 +23,7 @@ public class MsgHandler {
System.out.println("接收到消息：" + msg);
MqttProperties mqttProperties = MqttProperties.configBuild(
"47.102.133.88",
"mqtt/test"
"test1"
);
log.error("接收到消息初始化信息:{}",mqttProperties);
MqttClient mqttClient = mqttFactory.creatClient(mqttProperties);

52
src/main/java/com/muyu/receive/PublishSample.java 100644
View File

@ -0,0 +1,52 @@
package com.muyu.receive;
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;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class PublishSample {
public static void main(String[] args) {
String broker = "tcp://47.102.133.88:1883";
String topic = "test1";
String username = "";
String password = "";
String clientid = "publish_client";
String content = "Hello MQTT";
int qos = 0;
try {
MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
// 连接参数
MqttConnectOptions options = new MqttConnectOptions();
// 设置用户名和密码
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
// 连接
client.connect(options);
// 创建消息并设置 QoS
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(qos);
// 发布消息
client.publish(topic, message);
System.out.println("Message published");
System.out.println("topic: " + topic);
System.out.println("message content: " + content);
// 关闭连接
client.disconnect();
// 关闭客户端
client.close();
} catch (MqttException e) {
throw new RuntimeException(e);
}
SpringApplication.run(PublishSample.class, args);
}
}

60
src/main/java/com/muyu/receive/ReceiveCarMessage.java 100644
View File

@ -0,0 +1,60 @@
package com.muyu.receive;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* 接受小车报文
*
* @author Yangle
* Date 2024/5/31 20:26
*/
@SpringBootApplication
public class ReceiveCarMessage {
public static void main(String[] args) {
String broker ="tcp://47.102.133.88:1883";
String topic="mqtt/test";
String username="emqx";
String password="public";
String cliented="subscribe_client";
int qos=0;
try {
MqttClient client = new MqttClient(broker, cliented, new MemoryPersistence());
//连接参数
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
//设置回调
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
System.out.println("connectionLost:"+cause.getMessage());
}
@Override
public void messageArrived(String s, MqttMessage message) throws Exception {
System.out.println("topice:"+topic);
System.out.println("Qos:"+message.getQos());
System.out.println("message content:"+new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
});
client.connect(options);
client.subscribe(topic,qos);
} catch (MqttException e) {
throw new RuntimeException(e);
}
SpringApplication.run(ReceiveCarMessage.class, args);
}
}

49
src/main/java/com/muyu/receive/SubscribeSample.java 100644
View File

@ -0,0 +1,49 @@
package com.muyu.receive;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class SubscribeSample {
public static void main(String[] args) {
String broker = "tcp://47.102.133.88:1883";
String topic = "mqtt/test";
String username = "emqx";
String password = "public";
String clientid = "subscribe_client";
int qos = 0;
try {
MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
// 连接参数
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
// 设置回调
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
System.out.println("connectionLost: " + cause.getMessage());
}
@Override
public void messageArrived(String topic, MqttMessage message) {
System.out.println("topic: " + topic);
System.out.println("Qos: " + message.getQos());
System.out.println("message content: " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
});
client.connect(options);
client.subscribe(topic, qos);
} catch (Exception e) {
e.printStackTrace();
}
}
}