'更新mqtt支持'
This commit is contained in:
zy
parent
abaef5dfe4
commit
c7e837c0e7
|
|
@ -0,0 +1,13 @@
|
||||||
|
package org.jeecg.modules.mqtt;
|
||||||
|
|
||||||
|
import io.swagger.annotations.ApiModelProperty;
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
|
||||||
|
@Data
|
||||||
|
public class DTOMqtt {
|
||||||
|
@ApiModelProperty("主题")
|
||||||
|
private String topic;
|
||||||
|
@ApiModelProperty("发送的内容")
|
||||||
|
private String message;
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,34 @@
|
||||||
|
package org.jeecg.modules.mqtt;
|
||||||
|
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
import java.io.Serializable;
|
||||||
|
|
||||||
|
@Data
|
||||||
|
public class LhMqttMsg implements Serializable {
|
||||||
|
|
||||||
|
private static final long serialVersionUID = -8303548938481407659L;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* MD5值：MD5_lower(content + timestamp)
|
||||||
|
*/
|
||||||
|
private String md5;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息内容
|
||||||
|
*/
|
||||||
|
private String content = "";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息内容
|
||||||
|
*/
|
||||||
|
private byte[] byteContent;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 时间戳
|
||||||
|
*/
|
||||||
|
private Long timestamp;
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
@ -0,0 +1,71 @@
|
||||||
|
package org.jeecg.modules.mqtt;
|
||||||
|
|
||||||
|
import cn.hutool.core.lang.UUID;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.eclipse.paho.client.mqttv3.MqttException;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
import org.springframework.core.env.Environment;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
@Configuration
|
||||||
|
@ConditionalOnProperty(
|
||||||
|
prefix = "iot.mq",
|
||||||
|
name = "enabled",
|
||||||
|
havingValue = "true",
|
||||||
|
matchIfMissing = false // 默认注册，除非显式设置为 false
|
||||||
|
)
|
||||||
|
public class MqttConfig {
|
||||||
|
|
||||||
|
@Value("${mqtt.host}")
|
||||||
|
public String host;
|
||||||
|
@Value("${mqtt.username}")
|
||||||
|
public String username;
|
||||||
|
@Value("${mqtt.password}")
|
||||||
|
public String password;
|
||||||
|
// @Value("${mqtt.clientId}")
|
||||||
|
public String clientId = "jppApp-" + UUID.fastUUID().toString().replaceAll("-","").substring(0,5);
|
||||||
|
@Value("${mqtt.timeout}")
|
||||||
|
public int timeOut;
|
||||||
|
@Value("${mqtt.keepalive}")
|
||||||
|
public int keepAlive;
|
||||||
|
|
||||||
|
@Value("${mqtt.clearSession}")
|
||||||
|
public boolean clearSession;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private Environment env;
|
||||||
|
|
||||||
|
@Bean//注入Spring
|
||||||
|
public MyMqttClient myMqttClient() {
|
||||||
|
MyMqttClient myMqttClient = new MyMqttClient(host, username, password, clientId, timeOut, keepAlive, clearSession);
|
||||||
|
String activeProfile = env.getActiveProfiles()[0];
|
||||||
|
//正式环境才连接mqtt
|
||||||
|
if(activeProfile.contains("prod")){
|
||||||
|
log.warn("-------------正式环境,MQTT启动初始化---------");
|
||||||
|
for (int i = 0; i < 10; i++) {
|
||||||
|
try {
|
||||||
|
myMqttClient.connect();
|
||||||
|
// 这里可以订阅主题，推荐放到 MqttCallbackExtended.connectComplete方法中
|
||||||
|
//myMqttClient.subscribe("ABC", 1);
|
||||||
|
return myMqttClient;
|
||||||
|
} catch (MqttException e) {
|
||||||
|
log.error("== MqttConfig ==> MQTT connect exception, connect time = {}", i);
|
||||||
|
log.error("=========mqtt连接异常："+e.getMessage());
|
||||||
|
try {
|
||||||
|
Thread.sleep(2000);
|
||||||
|
} catch (InterruptedException e1) {
|
||||||
|
e1.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}else{
|
||||||
|
log.warn("-------------非正式环境跳过Mqtt订阅---------");
|
||||||
|
}
|
||||||
|
return myMqttClient;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,45 @@
|
||||||
|
package org.jeecg.modules.mqtt;
|
||||||
|
|
||||||
|
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
|
||||||
|
@ConditionalOnProperty(
|
||||||
|
prefix = "iot.mq",
|
||||||
|
name = "enabled",
|
||||||
|
havingValue = "true",
|
||||||
|
matchIfMissing = false // 默认注册，除非显式设置为 false
|
||||||
|
)
|
||||||
|
public interface MqttService {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 添加订阅主题
|
||||||
|
*
|
||||||
|
* @param topic 主题名称
|
||||||
|
*/
|
||||||
|
void addTopic(String topic);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 取消订阅主题
|
||||||
|
*
|
||||||
|
* @param topic 主题名称
|
||||||
|
*/
|
||||||
|
void removeTopic(String topic);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发布主题消息内容
|
||||||
|
*
|
||||||
|
* @param msgContent
|
||||||
|
* @param topic
|
||||||
|
*/
|
||||||
|
void publish(String topic,String msgContent);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发布主题消息内容
|
||||||
|
*
|
||||||
|
* @param msgContent
|
||||||
|
* @param topic
|
||||||
|
*/
|
||||||
|
void publish(String topic,byte[] msgContent);
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
@ -0,0 +1,170 @@
|
||||||
|
package org.jeecg.modules.mqtt;
|
||||||
|
|
||||||
|
import com.alibaba.fastjson.JSONObject;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
|
||||||
|
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
|
||||||
|
import org.eclipse.paho.client.mqttv3.MqttException;
|
||||||
|
import org.eclipse.paho.client.mqttv3.MqttMessage;
|
||||||
|
import org.jeecg.common.util.SpringContextUtils;
|
||||||
|
import org.jeecg.modules.config.RedisDistributedLock2;
|
||||||
|
import org.jeecg.modules.iot.enums.IotInerfaceTopicType;
|
||||||
|
import org.jeecg.modules.iot.o.dto.DTOMqttMessage;
|
||||||
|
import org.jeecg.modules.iot.o.mqtt.up.DeviceActionVo;
|
||||||
|
import org.jeecg.modules.iot.service.IFIotVirtualDeviceService;
|
||||||
|
import org.jeecg.modules.rocketmq.service.RocketMqService;
|
||||||
|
import org.springframework.data.redis.core.RedisTemplate;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Random;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
public class MyMqttCallback implements MqttCallbackExtended {
|
||||||
|
|
||||||
|
//手动注入
|
||||||
|
private MqttConfig mqttConfig = SpringContextUtils.getBean(MqttConfig.class);
|
||||||
|
|
||||||
|
private IFIotVirtualDeviceService deviceService = SpringContextUtils.getBean(IFIotVirtualDeviceService.class);
|
||||||
|
|
||||||
|
private static RedisTemplate redisTemplate = SpringContextUtils.getBean("redisTemplate", RedisTemplate.class);
|
||||||
|
|
||||||
|
private static RocketMqService rocketMqService = SpringContextUtils.getBean(RocketMqService.class);
|
||||||
|
|
||||||
|
private MyMqttClient myMqttClient;
|
||||||
|
|
||||||
|
public MyMqttCallback(MyMqttClient myMqttClient) {
|
||||||
|
this.myMqttClient = myMqttClient;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* MQTT Broker连接成功时被调用的方法。在该方法中可以执行 订阅系统约定的主题（推荐使用）。
|
||||||
|
* 如果 MQTT Broker断开连接之后又重新连接成功时，主题也需要再次订阅，将重新订阅主题放在连接成功后的回调方法中比较合理。
|
||||||
|
*
|
||||||
|
* @param reconnect
|
||||||
|
* @param serverURI MQTT Broker的url
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public void connectComplete(boolean reconnect, String serverURI) {
|
||||||
|
String connectMode = reconnect ? "重连" : "直连";
|
||||||
|
log.info("== MyMqttCallback ==> MQTT 连接成功，连接方式：{}，serverURI：{}", connectMode, serverURI);
|
||||||
|
//订阅主题
|
||||||
|
//查询所有需要订阅的mqtt主题
|
||||||
|
List<String> urlList = deviceService.getAllMqttTopic();
|
||||||
|
//去重
|
||||||
|
urlList = urlList.stream()
|
||||||
|
.distinct()
|
||||||
|
.collect(Collectors.toList());
|
||||||
|
for (String url : urlList) {//共享主题语法
|
||||||
|
myMqttClient.subscribe(url,1);
|
||||||
|
}
|
||||||
|
log.info("== MyMqttCallback ==> 连接方式：{}，订阅主题成功，topic：{}", connectMode, String.join(",", urlList));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 丢失连接，可在这里做重连
|
||||||
|
* 只会调用一次
|
||||||
|
*
|
||||||
|
* @param throwable
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public void connectionLost(Throwable throwable) {
|
||||||
|
log.error("== MyMqttCallback ==> connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());
|
||||||
|
long reconnectTimes = 1;
|
||||||
|
while (true) {
|
||||||
|
try {
|
||||||
|
if (MyMqttClient.getClient().isConnected()) {
|
||||||
|
//判断已经重新连接成功 需要重新订阅主题 可以在这个if里面订阅主题 或者 connectComplete（方法里面） 看你们自己选择
|
||||||
|
log.warn("== MyMqttCallback ==> mqtt reconnect success end 重新连接 重新订阅成功");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
reconnectTimes += 1;
|
||||||
|
log.warn("== MyMqttCallback ==> mqtt reconnect times = {} try again... mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
|
||||||
|
MyMqttClient.getClient().reconnect();
|
||||||
|
} catch (MqttException e) {
|
||||||
|
log.error("== MyMqttCallback ==> mqtt断连异常", e);
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
Thread.sleep(5000);
|
||||||
|
} catch (InterruptedException e1) {
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 接收到消息（subscribe订阅的主题消息）时被调用的方法
|
||||||
|
*
|
||||||
|
* @param topic
|
||||||
|
* @param mqttMessage
|
||||||
|
* @throws Exception 后得到的消息会执行到这里面
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
|
||||||
|
String o1 = new String(mqttMessage.getPayload());
|
||||||
|
log.info("== MyMqttCallback ==> messageArrived 接收消息主题: {}，接收消息内容: {}", topic, o1);
|
||||||
|
try {
|
||||||
|
if (topic.startsWith(IotInerfaceTopicType.DTU_TOPIC.getCode())) {
|
||||||
|
log.warn("fe-check=======================" + mqttMessage.getPayload());
|
||||||
|
//识别主题，确认设备
|
||||||
|
deviceService.processInterface(topic, mqttMessage.getPayload());
|
||||||
|
} else if (topic.startsWith(IotInerfaceTopicType.LH_IOT_TOPIC_UP.getCode())) {//蓝海虚拟设备上行逻辑
|
||||||
|
//暂时无业务
|
||||||
|
} else if (topic.contains(IotInerfaceTopicType.LH_IOT_TOPIC_DOWN.getCode())) {//蓝海虚拟设备下行逻辑，处理控制
|
||||||
|
// 尝试获取锁，等待3秒，重试间隔100ms
|
||||||
|
try {
|
||||||
|
// Boolean lock = RedisDistributedLock2.getLock("mqtt_receive_logic:__"+topic+"_"+o1.hashCode(),2);
|
||||||
|
// if(lock){
|
||||||
|
// System.out.println("锁测试" + o1 + " on topic: " + topic + " - " + lock);
|
||||||
|
// } else {
|
||||||
|
// System.out.println("锁测试" + o1 + " on topic: " + topic+ " - " + lock);
|
||||||
|
// return;
|
||||||
|
// }
|
||||||
|
log.warn("蓝海-check=======================" + mqttMessage.getPayload());
|
||||||
|
//解析数据获取请求id
|
||||||
|
String message = o1;
|
||||||
|
DeviceActionVo deviceQueryActionVo = JSONObject.toJavaObject(JSONObject.parseObject(message), DeviceActionVo.class);
|
||||||
|
//设备响应处理，查询、控制
|
||||||
|
deviceService.processIotAction(deviceQueryActionVo);
|
||||||
|
//MQ模式
|
||||||
|
// DTOMqttMessage dtoMqttMessage = new DTOMqttMessage();
|
||||||
|
// dtoMqttMessage.setTopic(topic);
|
||||||
|
// dtoMqttMessage.setMessage(message);
|
||||||
|
// rocketMqService.sendMqttMessage(dtoMqttMessage);
|
||||||
|
}catch (Exception e){
|
||||||
|
log.error("======处理蓝海主题设备报错================"+e.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}catch (Exception e){
|
||||||
|
e.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息发送（publish）完成时被调用的方法
|
||||||
|
*
|
||||||
|
* @param iMqttDeliveryToken
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
|
||||||
|
log.info("== MyMqttCallback ==> deliveryComplete 消息发送完成，Complete= {}", iMqttDeliveryToken.isComplete());
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
// 指定最小和最大范围
|
||||||
|
public static void randomSleep(int minMs, int maxMs) {
|
||||||
|
try {
|
||||||
|
Random random = new Random();
|
||||||
|
if (minMs >= maxMs) {
|
||||||
|
throw new IllegalArgumentException("最小值必须小于最大值");
|
||||||
|
}
|
||||||
|
int sleepTime = minMs + random.nextInt(maxMs - minMs + 1);
|
||||||
|
System.out.println("随机休眠: " + sleepTime + "ms");
|
||||||
|
Thread.sleep(sleepTime);
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
@ -0,0 +1,209 @@
|
||||||
|
package org.jeecg.modules.mqtt;
|
||||||
|
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.eclipse.paho.client.mqttv3.*;
|
||||||
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
|
public class MyMqttClient {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* MQTT Broker 基本连接参数，用户名、密码为非必选参数
|
||||||
|
*/
|
||||||
|
private String host;
|
||||||
|
private String username;
|
||||||
|
private String password;
|
||||||
|
private String clientId;
|
||||||
|
private int timeout;
|
||||||
|
private int keepalive;
|
||||||
|
private boolean clearSession;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* MQTT 客户端
|
||||||
|
*/
|
||||||
|
private static MqttClient client;
|
||||||
|
|
||||||
|
public static MqttClient getClient() {
|
||||||
|
return client;
|
||||||
|
}
|
||||||
|
|
||||||
|
public static void setClient(MqttClient client) {
|
||||||
|
MyMqttClient.client = client;
|
||||||
|
}
|
||||||
|
|
||||||
|
public MyMqttClient(String host, String username, String password, String clientId, int timeOut, int keepAlive, boolean clearSession) {
|
||||||
|
this.host = host;
|
||||||
|
this.username = username;
|
||||||
|
this.password = password;
|
||||||
|
this.clientId = clientId;
|
||||||
|
this.timeout = timeOut;
|
||||||
|
this.keepalive = keepAlive;
|
||||||
|
this.clearSession = clearSession;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 设置 MQTT Broker 基本连接参数
|
||||||
|
*
|
||||||
|
* @param username
|
||||||
|
* @param password
|
||||||
|
* @param timeout
|
||||||
|
* @param keepalive
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
|
public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {
|
||||||
|
MqttConnectOptions options = new MqttConnectOptions();
|
||||||
|
options.setUserName(username);
|
||||||
|
options.setPassword(password.toCharArray());
|
||||||
|
options.setConnectionTimeout(timeout);
|
||||||
|
options.setKeepAliveInterval(keepalive);
|
||||||
|
options.setCleanSession(clearSession);
|
||||||
|
options.setAutomaticReconnect(true);
|
||||||
|
return options;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 连接 MQTT Broker，得到 MqttClient连接对象
|
||||||
|
*/
|
||||||
|
public void connect() throws MqttException {
|
||||||
|
if (client == null) {
|
||||||
|
client = new MqttClient(host, clientId, new MemoryPersistence());
|
||||||
|
// 设置回调
|
||||||
|
client.setCallback(new MyMqttCallback(MyMqttClient.this));
|
||||||
|
}
|
||||||
|
// 连接参数
|
||||||
|
MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);
|
||||||
|
if (!client.isConnected()) {
|
||||||
|
client.connect(mqttConnectOptions);
|
||||||
|
} else {
|
||||||
|
client.disconnect();
|
||||||
|
client.connect(mqttConnectOptions);
|
||||||
|
}
|
||||||
|
log.info("== MyMqttClient ==> MQTT connect success");//未发生异常，则连接成功
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发布，默认qos为0，非持久化
|
||||||
|
*
|
||||||
|
* @param pushMessage
|
||||||
|
* @param topic
|
||||||
|
*/
|
||||||
|
public void publish(String pushMessage, String topic) {
|
||||||
|
publish(pushMessage, topic, 2, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发布，默认qos为0，非持久化
|
||||||
|
*
|
||||||
|
* @param pushMessage
|
||||||
|
* @param topic
|
||||||
|
*/
|
||||||
|
public void publish(byte[] pushMessage, String topic) {
|
||||||
|
publish(pushMessage, topic, 2, false);//至少一次
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发布消息
|
||||||
|
*
|
||||||
|
* @param pushMessage
|
||||||
|
* @param topic
|
||||||
|
* @param qos
|
||||||
|
* @param retained:留存
|
||||||
|
*/
|
||||||
|
public void publish(String pushMessage, String topic, int qos, boolean retained) {
|
||||||
|
MqttMessage message = new MqttMessage();
|
||||||
|
message.setPayload(pushMessage.getBytes());
|
||||||
|
message.setQos(qos);
|
||||||
|
message.setRetained(retained);
|
||||||
|
MqttTopic mqttTopic = MyMqttClient.getClient().getTopic(topic);
|
||||||
|
if (null == mqttTopic) {
|
||||||
|
log.error("== MyMqttClient ==> topic is not exist");
|
||||||
|
}
|
||||||
|
MqttDeliveryToken token;//Delivery:配送
|
||||||
|
synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
|
||||||
|
try {
|
||||||
|
token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
|
||||||
|
token.waitForCompletion(3000L);
|
||||||
|
} catch (MqttPersistenceException e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
} catch (MqttException e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发布消息
|
||||||
|
*
|
||||||
|
* @param pushMessage
|
||||||
|
* @param topic
|
||||||
|
* @param qos
|
||||||
|
* @param retained:留存
|
||||||
|
*/
|
||||||
|
public void publish(byte[] pushMessage, String topic, int qos, boolean retained) {
|
||||||
|
MqttMessage message = new MqttMessage();
|
||||||
|
message.setPayload(pushMessage);
|
||||||
|
message.setQos(qos);
|
||||||
|
message.setRetained(retained);
|
||||||
|
MqttTopic mqttTopic = MyMqttClient.getClient().getTopic(topic);
|
||||||
|
if (null == mqttTopic) {
|
||||||
|
log.error("== byte MyMqttClient ==> topic is not exist");
|
||||||
|
}
|
||||||
|
MqttDeliveryToken token;//Delivery:配送
|
||||||
|
synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
|
||||||
|
try {
|
||||||
|
token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
|
||||||
|
token.waitForCompletion(1000L);
|
||||||
|
} catch (MqttPersistenceException e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
} catch (MqttException e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 订阅某个主题，qos默认为0
|
||||||
|
*
|
||||||
|
* @param topic
|
||||||
|
*/
|
||||||
|
public void subscribe(String topic) {
|
||||||
|
subscribe(topic, 2);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 订阅某个主题
|
||||||
|
*
|
||||||
|
* @param topic
|
||||||
|
* @param qos
|
||||||
|
*/
|
||||||
|
public void subscribe(String topic, int qos) {
|
||||||
|
try {
|
||||||
|
MyMqttClient.getClient().subscribe(topic, qos);
|
||||||
|
} catch (MqttException e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
}
|
||||||
|
log.info("== MyMqttClient ==> 订阅主题成功：topic = {}， qos = {}", topic, qos);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 取消订阅主题
|
||||||
|
*
|
||||||
|
* @param topic 主题名称
|
||||||
|
*/
|
||||||
|
public void cleanTopic(String topic) {
|
||||||
|
if (client != null && client.isConnected()) {
|
||||||
|
try {
|
||||||
|
client.unsubscribe(topic);
|
||||||
|
} catch (MqttException e) {
|
||||||
|
e.printStackTrace();
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
log.error("== MyMqttClient ==> 取消订阅失败！");
|
||||||
|
}
|
||||||
|
log.info("== MyMqttClient ==> 取消订阅主题成功：topic = {}", topic);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
@ -0,0 +1,37 @@
|
||||||
|
package org.jeecg.modules.mqtt;
|
||||||
|
|
||||||
|
import io.swagger.annotations.Api;
|
||||||
|
import io.swagger.annotations.ApiOperation;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.PostMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
@RestController
|
||||||
|
@RequestMapping("/mqtt")
|
||||||
|
@Api(value = "MyMqttController", tags = {"MQTT相关操作接口"})
|
||||||
|
public class MyMqttController {
|
||||||
|
@Autowired(required = false)
|
||||||
|
private MqttService mqttService;
|
||||||
|
|
||||||
|
@GetMapping("/addTopic")
|
||||||
|
@ApiOperation(value = "添加订阅主题接口")
|
||||||
|
public void addTopic(String topic) {
|
||||||
|
mqttService.addTopic(topic);
|
||||||
|
}
|
||||||
|
|
||||||
|
@GetMapping("/removeTopic")
|
||||||
|
@ApiOperation(value = "取消订阅主题接口")
|
||||||
|
public void removeTopic(String topic) {
|
||||||
|
mqttService.removeTopic(topic);
|
||||||
|
}
|
||||||
|
|
||||||
|
@PostMapping("/removeTopic")
|
||||||
|
@ApiOperation(value = "发布主题消息内容接口")
|
||||||
|
public void removeTopic(String msgContent, String topic) {
|
||||||
|
mqttService.publish(msgContent, topic);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
|
@ -0,0 +1,52 @@
|
||||||
|
package org.jeecg.modules.mqtt.impl;
|
||||||
|
|
||||||
|
import cn.hutool.core.lang.UUID;
|
||||||
|
import com.alibaba.fastjson.JSON;
|
||||||
|
import org.jeecg.modules.mqtt.MqttService;
|
||||||
|
import org.jeecg.modules.mqtt.MyMqttClient;
|
||||||
|
import org.jeecg.modules.mqtt.LhMqttMsg;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
@ConditionalOnProperty(
|
||||||
|
prefix = "iot.mq",
|
||||||
|
name = "enabled",
|
||||||
|
havingValue = "true",
|
||||||
|
matchIfMissing = false // 默认注册，除非显式设置为 false
|
||||||
|
)
|
||||||
|
@Service
|
||||||
|
public class MqttServiceImpl implements MqttService {
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private MyMqttClient myMqttClient;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void addTopic(String topic) {
|
||||||
|
myMqttClient.subscribe(topic);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void removeTopic(String topic) {
|
||||||
|
myMqttClient.cleanTopic(topic);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void publish(String topic,String msgContent) {
|
||||||
|
//MyXxxMqttMsg 转Json
|
||||||
|
// LhMqttMsg myXxxMqttMsg = new LhMqttMsg();
|
||||||
|
// myXxxMqttMsg.setContent(msgContent);
|
||||||
|
// myXxxMqttMsg.setTimestamp(System.currentTimeMillis());
|
||||||
|
// // TODO Md5值
|
||||||
|
// myXxxMqttMsg.setMd5(UUID.randomUUID().toString());
|
||||||
|
// String msgJson = JSON.toJSONString(myXxxMqttMsg);
|
||||||
|
|
||||||
|
//发布消息
|
||||||
|
myMqttClient.publish(msgContent, topic);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void publish(String topic,byte[] msgContent) {
|
||||||
|
myMqttClient.publish(msgContent, topic);
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue