Administrator
/
FenXiNspJavaGather
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

增加多Mqtt实例

This commit is contained in:
zhangyue 2026-07-29 16:28:09 +08:00
parent 246483f4ec
commit 340abe8a18
21 changed files with 954 additions and 328 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/lanhai/constant/IotConstants.java
View File

@ -221,4 +221,10 @@ public interface IotConstants {
*/
String lhviot_standard = "lhviot_standard";
/**
* mqtt连接协议
*/
String CONFIG_TYPE_MQTT = "mqtt";
}

6
src/main/java/com/lanhai/entity/SurvConfig.java
View File

@ -88,6 +88,12 @@ public class SurvConfig extends Model<SurvConfig> {
@TableField("CONFIG_TYPE")
private String configType;
/**
* 系统内置
*/
@TableField("CONFIG_SUB_TYPE")
private String configSubType;
/**
* 备注
*/

1
src/main/java/com/lanhai/enums/IotInerfaceTopicType.java
View File

@ -9,6 +9,7 @@ public enum IotInerfaceTopicType {
LH_IOT_TOPIC_UP("/lh-iot-up","蓝海虚拟设备头上行"),
LH_IOT_TOPIC_DOWN("/lh-iot-down","蓝海虚拟设备头下行"),
API_TOPIC("/api-","厂家api头"),
RENKE_TOPIC("rtd/","厂家仁科头"),
OTHER("other","其他")
;

229
src/main/java/com/lanhai/mqtt/config/MqttClientManager.java Normal file
View File

@ -0,0 +1,229 @@
package com.lanhai.mqtt.config;
import com.lanhai.mqtt.handler.MqttCallbackHandler;
import com.lanhai.mqtt.service.MessageProcessor;
import com.lanhai.service.ISurvIotVirtualDeviceService;
import lombok.extern.slf4j.Slf4j;
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.persist.MemoryPersistence;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@Slf4j
@Component
public class MqttClientManager {
private final MqttProperties mqttProperties;
private final ISurvIotVirtualDeviceService deviceService;
private final MessageProcessor messageProcessor;
/**
* 存储所有MQTT客户端实例，key为客户端名称
*/
private final Map<String, MqttClient> clientMap = new ConcurrentHashMap<>();
/**
* 存储每个客户端对应的回调处理器
*/
private final Map<String, MqttCallbackHandler> callbackMap = new ConcurrentHashMap<>();
public MqttClientManager(MqttProperties mqttProperties,
ISurvIotVirtualDeviceService deviceService,
MessageProcessor messageProcessor) {
this.mqttProperties = mqttProperties;
this.deviceService = deviceService;
this.messageProcessor = messageProcessor;
}
/**
* 初始化所有MQTT连接
*/
public void init() {
if (!Boolean.TRUE.equals(mqttProperties.getEnabled())) {
log.info("MQTT 未启用，跳过初始化");
return;
}
List<MqttProperties.MqttClientConfig> clients = mqttProperties.getClients();
if (clients == null || clients.isEmpty()) {
log.warn("MQTT 客户端配置列表为空");
return;
}
for (MqttProperties.MqttClientConfig config : clients) {
if (Boolean.FALSE.equals(config.getEnabled())) {
log.info("MQTT 客户端 [{}] 已禁用，跳过", config.getName());
continue;
}
try {
initClient(config);
} catch (Exception e) {
log.error("MQTT 客户端 [{}] 初始化失败", config.getName(), e);
}
}
}
/**
* 初始化单个MQTT客户端
*/
private void initClient(MqttProperties.MqttClientConfig config) throws MqttException {
String clientName = config.getName();
log.info("========== 初始化 MQTT 客户端 [{}] ==========", clientName);
// 创建客户端
MqttClient mqttClient = new MqttClient(
config.getBroker().getUrl(),
config.getClient().getId(),
new MemoryPersistence()
);
// 创建对应的回调处理器
MqttCallbackHandler callbackHandler = new MqttCallbackHandler();
callbackHandler.setClientName(clientName);
callbackHandler.setMqttClient(mqttClient);
callbackHandler.setMessageProcessor(messageProcessor);
mqttClient.setCallback(callbackHandler);
// 配置连接选项
MqttConnectOptions options = createConnectOptions(config);
// 连接
mqttClient.connect(options);
log.info("MQTT 客户端 [{}] 连接成功！", clientName);
// 存储到Map
clientMap.put(clientName, mqttClient);
callbackMap.put(clientName, callbackHandler);
// 订阅主题
subscribeTopics(clientName, mqttClient, config);
}
/**
* 创建连接选项
*/
private MqttConnectOptions createConnectOptions(MqttProperties.MqttClientConfig config) {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(config.getClient().getCleanSession());
options.setConnectionTimeout(config.getConnection().getTimeout());
options.setKeepAliveInterval(config.getConnection().getKeepAlive());
if (config.getConnection().getAutomaticReconnect() != null) {
options.setAutomaticReconnect(config.getConnection().getAutomaticReconnect());
}
if (config.getAuth().getUsername() != null) {
options.setUserName(config.getAuth().getUsername());
options.setPassword(config.getAuth().getPassword().toCharArray());
}
return options;
}
/**
* 订阅主题
*/
private void subscribeTopics(String clientName, MqttClient mqttClient,
MqttProperties.MqttClientConfig config) {
try {
// 查询所有需要订阅的mqtt主题
List<String> urlList = deviceService.getAllMqttTopic(clientName);
urlList = urlList.stream()
.distinct()
.collect(Collectors.toList());
for (String url : urlList) {
mqttClient.subscribe(url, config.getQos());
}
log.info("MQTT 客户端 [{}] 订阅主题成功，topic：{}",
clientName, String.join(",", urlList));
} catch (MqttException e) {
log.error("MQTT 客户端 [{}] 订阅主题失败", clientName, e);
}
}
/**
* 获取指定名称的MQTT客户端
*/
public MqttClient getClient(String name) {
return clientMap.get(name);
}
/**
* 获取默认MQTT客户端
*/
public MqttClient getDefaultClient() {
String defaultClient = mqttProperties.getDefaultClient();
if (defaultClient != null && clientMap.containsKey(defaultClient)) {
return clientMap.get(defaultClient);
}
// 如果没有指定默认客户端，返回第一个
if (!clientMap.isEmpty()) {
return clientMap.values().iterator().next();
}
return null;
}
/**
* 获取所有客户端名称
*/
public List<String> getAllClientNames() {
return Collections.unmodifiableList(new ArrayList<>(clientMap.keySet()));
}
/**
* 检查指定客户端是否连接
*/
public boolean isConnected(String name) {
MqttClient client = clientMap.get(name);
return client != null && client.isConnected();
}
/**
* 检查默认客户端是否连接
*/
public boolean isDefaultConnected() {
MqttClient client = getDefaultClient();
return client != null && client.isConnected();
}
/**
* 获取指定客户端的配置
*/
public MqttProperties.MqttClientConfig getClientConfig(String name) {
return mqttProperties.getClients().stream()
.filter(c -> name.equals(c.getName()))
.findFirst()
.orElse(null);
}
/**
* 断开所有连接
*/
@PreDestroy
public void destroy() {
for (Map.Entry<String, MqttClient> entry : clientMap.entrySet()) {
MqttClient client = entry.getValue();
if (client != null && client.isConnected()) {
try {
client.disconnect();
log.info("MQTT 客户端 [{}] 已断开连接", entry.getKey());
} catch (MqttException e) {
log.error("断开 MQTT 客户端 [{}] 连接失败", entry.getKey(), e);
}
}
}
clientMap.clear();
callbackMap.clear();
}
}

135
src/main/java/com/lanhai/mqtt/config/MqttConfig.java
View File

@ -1,23 +1,12 @@
package com.lanhai.mqtt.config;
import com.lanhai.mqtt.handler.MqttCallbackHandler;
import com.lanhai.service.ISurvIotVirtualDeviceService;
import com.lanhai.service.Impl.SurvIotVirtualDeviceServiceImpl;
import com.lanhai.util.SpringContextUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.stream.Collectors;
import javax.annotation.PostConstruct;
@Slf4j
@Configuration
@ -25,123 +14,11 @@ import java.util.stream.Collectors;
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
public class MqttConfig {
private final MqttProperties mqttProperties;
private final MqttCallbackHandler mqttCallbackHandler; // 这里注入
private MqttClient mqttClient;
private final ISurvIotVirtualDeviceService deviceService;
private final MqttClientManager mqttClientManager;
@Bean
public MqttClient mqttClient() {
try {
log.info("========== 初始化 MQTT 客户端 ==========");
// 1. 创建客户端
mqttClient = new MqttClient(
mqttProperties.getBroker().getUrl(),
mqttProperties.getClient().getId(),
new MemoryPersistence()
);
// 2. ⭐ 将 MqttClient 注入到 CallbackHandler
mqttCallbackHandler.setMqttClient(mqttClient);
// 3. 设置回调
mqttClient.setCallback(mqttCallbackHandler);
// 4. 配置连接选项
MqttConnectOptions options = createConnectOptions();
// 5. 连接
mqttClient.connect(options);
log.info("MQTT 连接成功！");
// 6. 订阅主题
subscribeTopics();
return mqttClient;
} catch (MqttException e) {
log.error("MQTT 连接失败", e);
throw new RuntimeException("MQTT 连接失败", e);
}
}
private void subscribeTopics() {
try {
// // 订阅接收主题
// if (mqttProperties.getTopics().getReceive() != null) {
// mqttClient.subscribe(
// mqttProperties.getTopics().getReceive(),
// mqttProperties.getQos()
// );
// log.info("订阅主题: {}", mqttProperties.getTopics().getReceive());
// }
//
// // 订阅状态主题
// if (mqttProperties.getTopics().getStatus() != null) {
// mqttClient.subscribe(
// mqttProperties.getTopics().getStatus(),
// mqttProperties.getQos()
// );
// log.info("订阅主题: {}", mqttProperties.getTopics().getStatus());
// }
//
// // 订阅设备通配符主题
// if (mqttProperties.getTopics().getDevice() != null) {
// mqttClient.subscribe(
// mqttProperties.getTopics().getDevice(),
// mqttProperties.getQos()
// );
// log.info("订阅通配符主题: {}", mqttProperties.getTopics().getDevice());
// }
//查询所有需要订阅的mqtt主题
List<String> urlList = deviceService.getAllMqttTopic();
//去重
urlList = urlList.stream()
.distinct()
.collect(Collectors.toList());
for (String url : urlList) {//共享主题语法
mqttClient.subscribe(
url,
mqttProperties.getQos()
);
}
log.error("== MyMqttCallback ==> ，订阅主题成功，topic：{}", String.join(",", urlList));
} catch (MqttException e) {
log.error("订阅主题失败", e);
}
}
private MqttConnectOptions createConnectOptions() {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(mqttProperties.getClient().getCleanSession());
options.setConnectionTimeout(mqttProperties.getConnection().getTimeout());
options.setKeepAliveInterval(mqttProperties.getConnection().getKeepAlive());
if (mqttProperties.getConnection().getAutomaticReconnect() != null) {
options.setAutomaticReconnect(mqttProperties.getConnection().getAutomaticReconnect());
}
if (mqttProperties.getAuth().getUsername() != null) {
options.setUserName(mqttProperties.getAuth().getUsername());
options.setPassword(mqttProperties.getAuth().getPassword().toCharArray());
}
return options;
}
@PreDestroy
public void destroy() {
if (mqttClient != null && mqttClient.isConnected()) {
try {
mqttClient.disconnect();
log.info("MQTT 客户端已断开连接");
} catch (MqttException e) {
log.error("断开 MQTT 连接失败", e);
}
}
@PostConstruct
public void init() {
log.info("========== MQTT 配置初始化 ==========");
mqttClientManager.init();
}
}

48
src/main/java/com/lanhai/mqtt/config/MqttProperties.java
View File

@ -4,18 +4,62 @@ import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
@Data
@Component
@ConfigurationProperties(prefix = "mqtt")
public class MqttProperties {
private Boolean enabled;
/**
* 默认客户端（向后兼容单连接配置）
*/
private String defaultClient;
/**
* 多客户端配置列表
*/
private List<MqttClientConfig> clients = new ArrayList<>();
@Data
public static class MqttClientConfig {
/**
* 客户端名称，用于区分不同连接
*/
private String name;
/**
* 是否启用
*/
private Boolean enabled = true;
/**
* Broker配置
*/
private Broker broker = new Broker();
/**
* Client配置
*/
private Client client = new Client();
/**
* Connection配置
*/
private Connection connection = new Connection();
/**
* Auth配置
*/
private Auth auth = new Auth();
/**
* Topics配置
*/
private Topics topics = new Topics();
private Integer qos;
private Boolean retained;
/**
* QoS
*/
private Integer qos = 1;
/**
* Retained
*/
private Boolean retained = false;
}
@Data
public static class Broker {

38
src/main/java/com/lanhai/mqtt/handler/MqttCallbackHandler.java
View File

@ -6,27 +6,33 @@ import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
@Slf4j
@Component
public class MqttCallbackHandler implements MqttCallback {
private final MessageProcessor messageProcessor;
private MqttClient mqttClient; // 改为 Setter 注入
private MessageProcessor messageProcessor;
private MqttClient mqttClient;
private String clientName;
public MqttCallbackHandler() {
}
public MqttCallbackHandler(MessageProcessor messageProcessor) {
this.messageProcessor = messageProcessor;
}
/**
* 通过 Setter 注入 MqttClient，避免循环依赖
*/
public void setMessageProcessor(MessageProcessor messageProcessor) {
this.messageProcessor = messageProcessor;
}
public void setMqttClient(MqttClient mqttClient) {
this.mqttClient = mqttClient;
log.info("MqttClient 已注入到 CallbackHandler");
}
public void setClientName(String clientName) {
this.clientName = clientName;
}
@PostConstruct
@ -36,26 +42,30 @@ public class MqttCallbackHandler implements MqttCallback {
@Override
public void connectionLost(Throwable cause) {
log.error("MQTT 连接丢失，原因: {}", cause.getMessage(), cause);
log.error("MQTT 客户端 [{}] 连接丢失，原因: {}",
clientName, cause.getMessage(), cause);
if (mqttClient != null && !mqttClient.isConnected()) {
log.info("等待客户端自动重连...");
log.info("MQTT 客户端 [{}] 等待自动重连...", clientName);
}
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
String payload = new String(message.getPayload());
log.info("收到消息 - 主题: {}, QoS: {}", topic, message.getQos());
log.info("MQTT 客户端 [{}] 收到消息 - 主题: {}, QoS: {}",
clientName, topic, message.getQos());
// 异步处理消息
messageProcessor.processAsync(topic, payload);
if (messageProcessor != null) {
messageProcessor.processAsync(clientName,topic, payload);
}
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
try {
log.debug("消息发送完成 - Message ID: {}", token.getMessageId());
log.debug("MQTT 客户端 [{}] 消息发送完成 - Message ID: {}",
clientName, token.getMessageId());
} catch (Exception e) {
log.error("处理消息发送完成回调失败", e);
}

152
src/main/java/com/lanhai/mqtt/listener/MqttReconnectListener.java
View File

@ -1,6 +1,8 @@
package com.lanhai.mqtt.listener;
import com.lanhai.mqtt.config.MqttClientManager;
import com.lanhai.mqtt.config.MqttProperties;
import com.lanhai.service.ISurvIotVirtualDeviceService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
@ -15,6 +17,9 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@Slf4j
@ -24,11 +29,14 @@ import java.util.concurrent.atomic.AtomicBoolean;
@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true")
public class MqttReconnectListener {
private final MqttClient mqttClient;
private final MqttClientManager mqttClientManager;
private final MqttProperties mqttProperties;
private final ISurvIotVirtualDeviceService deviceService;
// 每个客户端独立的重连状态
private final Map<String, AtomicBoolean> reconnectingMap = new ConcurrentHashMap<>();
private final Map<String, Integer> reconnectCountMap = new ConcurrentHashMap<>();
private AtomicBoolean isReconnecting = new AtomicBoolean(false);
private int reconnectCount = 0;
private static final int MAX_RECONNECT_COUNT = 5;
private static final long RECONNECT_INTERVAL = 30000; // 30秒
@ -39,140 +47,160 @@ public class MqttReconnectListener {
@EventListener(ContextRefreshedEvent.class)
public void onApplicationReady() {
log.info("应用启动完成，MQTT 连接状态: {}", mqttClient.isConnected());
List<String> clientNames = mqttClientManager.getAllClientNames();
for (String name : clientNames) {
log.info("MQTT 客户端 [{}] 连接状态: {}", name, mqttClientManager.isConnected(name));
}
}
/**
* 定时检查连接状态并重连
* 定时检查所有客户端连接状态并重连
*/
@Scheduled(fixedDelay = 10000) // 每10秒检查一次
public void checkAndReconnect() {
List<String> clientNames = mqttClientManager.getAllClientNames();
for (String clientName : clientNames) {
MqttClient client = mqttClientManager.getClient(clientName);
if (client == null) continue;
// 如果已经连接，重置重连计数
if (mqttClient.isConnected()) {
if (reconnectCount > 0) {
log.info("MQTT 连接已恢复，重置重连计数器");
reconnectCount = 0;
isReconnecting.set(false);
if (client.isConnected()) {
Integer count = reconnectCountMap.get(clientName);
if (count != null && count > 0) {
log.info("MQTT 客户端 [{}] 连接已恢复，重置重连计数器", clientName);
reconnectCountMap.put(clientName, 0);
reconnectingMap.put(clientName, new AtomicBoolean(false));
}
return;
continue;
}
// 如果正在重连，跳过
AtomicBoolean isReconnecting = reconnectingMap.computeIfAbsent(clientName, k -> new AtomicBoolean(false));
if (isReconnecting.get()) {
return;
continue;
}
// 检查重连次数
int reconnectCount = reconnectCountMap.getOrDefault(clientName, 0);
if (reconnectCount >= MAX_RECONNECT_COUNT) {
log.error("MQTT 重连失败次数已达上限 ({})，停止重连", MAX_RECONNECT_COUNT);
return;
log.error("MQTT 客户端 [{}] 重连失败次数已达上限 ({})，停止重连", clientName, MAX_RECONNECT_COUNT);
continue;
}
// 执行重连
attemptReconnect();
attemptReconnect(clientName);
}
}
@Async
public void attemptReconnect() {
public void attemptReconnect(String clientName) {
AtomicBoolean isReconnecting = reconnectingMap.computeIfAbsent(clientName, k -> new AtomicBoolean(false));
if (!isReconnecting.compareAndSet(false, true)) {
return;
}
try {
reconnectCount++;
log.info("尝试重连 MQTT (第 {}/{} 次)...", reconnectCount, MAX_RECONNECT_COUNT);
int reconnectCount = reconnectCountMap.getOrDefault(clientName, 0) + 1;
reconnectCountMap.put(clientName, reconnectCount);
log.info("尝试重连 MQTT 客户端 [{}] (第 {}/{} 次)...", clientName, reconnectCount, MAX_RECONNECT_COUNT);
// 等待一段时间再重连
Thread.sleep(RECONNECT_INTERVAL);
// 重新连接
if (!mqttClient.isConnected()) {
MqttConnectOptions options = createConnectOptions();
mqttClient.connect(options);
log.info("MQTT 重连成功！");
MqttClient client = mqttClientManager.getClient(clientName);
if (client != null && !client.isConnected()) {
MqttProperties.MqttClientConfig config = mqttClientManager.getClientConfig(clientName);
if (config == null) {
log.error("MQTT 客户端 [{}] 配置不存在", clientName);
return;
}
MqttConnectOptions options = createConnectOptions(config);
client.connect(options);
log.info("MQTT 客户端 [{}] 重连成功！", clientName);
// 重连成功后重新订阅主题
resubscribeTopics();
resubscribeTopics(clientName, client, config);
// 重置重连计数
reconnectCount = 0;
reconnectCountMap.put(clientName, 0);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.warn("重连被中断");
log.warn("MQTT 客户端 [{}] 重连被中断", clientName);
} catch (MqttException e) {
log.error("MQTT 重连失败", e);
log.error("MQTT 客户端 [{}] 重连失败", clientName, e);
// 如果重连次数达到上限，发送告警
int reconnectCount = reconnectCountMap.getOrDefault(clientName, 0);
if (reconnectCount >= MAX_RECONNECT_COUNT) {
sendReconnectAlert();
sendReconnectAlert(clientName);
}
} finally {
isReconnecting.set(false);
}
}
private MqttConnectOptions createConnectOptions() {
private MqttConnectOptions createConnectOptions(MqttProperties.MqttClientConfig config) {
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(mqttProperties.getClient().getCleanSession());
options.setConnectionTimeout(mqttProperties.getConnection().getTimeout());
options.setKeepAliveInterval(mqttProperties.getConnection().getKeepAlive());
options.setCleanSession(config.getClient().getCleanSession());
options.setConnectionTimeout(config.getConnection().getTimeout());
options.setKeepAliveInterval(config.getConnection().getKeepAlive());
if (mqttProperties.getAuth().getUsername() != null) {
options.setUserName(mqttProperties.getAuth().getUsername());
options.setPassword(mqttProperties.getAuth().getPassword().toCharArray());
if (config.getConnection().getAutomaticReconnect() != null) {
options.setAutomaticReconnect(config.getConnection().getAutomaticReconnect());
}
if (config.getAuth().getUsername() != null) {
options.setUserName(config.getAuth().getUsername());
options.setPassword(config.getAuth().getPassword().toCharArray());
}
return options;
}
private void resubscribeTopics() {
private void resubscribeTopics(String clientName, MqttClient client, MqttProperties.MqttClientConfig config) {
try {
log.info("重新订阅主题...");
log.info("MQTT 客户端 [{}] 重新订阅主题...", clientName);
if (mqttProperties.getTopics().getReceive() != null) {
mqttClient.subscribe(mqttProperties.getTopics().getReceive(),
mqttProperties.getQos());
log.info("重新订阅: {}", mqttProperties.getTopics().getReceive());
}
// 查询所有需要订阅的主题
List<String> urlList = deviceService.getAllMqttTopic(clientName);
urlList = urlList.stream().distinct().collect(java.util.stream.Collectors.toList());
if (mqttProperties.getTopics().getStatus() != null) {
mqttClient.subscribe(mqttProperties.getTopics().getStatus(),
mqttProperties.getQos());
log.info("重新订阅: {}", mqttProperties.getTopics().getStatus());
}
if (mqttProperties.getTopics().getDevice() != null) {
mqttClient.subscribe(mqttProperties.getTopics().getDevice(),
mqttProperties.getQos());
log.info("重新订阅: {}", mqttProperties.getTopics().getDevice());
for (String url : urlList) {
client.subscribe(url, config.getQos());
}
// 发送上线消息
if (mqttProperties.getTopics().getStatus() != null) {
mqttClient.publish(
mqttProperties.getTopics().getStatus(),
if (config.getTopics() != null && config.getTopics().getStatus() != null) {
client.publish(
config.getTopics().getStatus(),
"online".getBytes(),
mqttProperties.getQos(),
config.getQos(),
true
);
log.info("已发送上线状态");
log.info("MQTT 客户端 [{}] 已发送上线状态", clientName);
}
log.info("MQTT 客户端 [{}] 重新订阅主题成功，topic：{}", clientName, String.join(",", urlList));
} catch (MqttException e) {
log.error("重新订阅主题失败", e);
log.error("MQTT 客户端 [{}] 重新订阅主题失败", clientName, e);
}
}
private void sendReconnectAlert() {
private void sendReconnectAlert(String clientName) {
MqttProperties.MqttClientConfig config = mqttClientManager.getClientConfig(clientName);
log.error("========== MQTT 重连告警 ==========");
log.error("MQTT 服务器: {}", mqttProperties.getBroker().getUrl());
log.error("MQTT 客户端: {}", clientName);
if (config != null) {
log.error("MQTT 服务器: {}", config.getBroker().getUrl());
}
log.error("重连失败次数: {}", MAX_RECONNECT_COUNT);
log.error("请检查 MQTT 服务器状态和网络连接");
log.error("==================================");
// 这里可以添加告警通知，如发送邮件、短信等
// emailService.sendAlert("MQTT连接失败");
// emailService.sendAlert("MQTT连接失败: " + clientName);
}
}

36
src/main/java/com/lanhai/mqtt/service/MessageProcessor.java
View File

@ -25,19 +25,19 @@ public class MessageProcessor {
/**
* 同步处理消息
*/
public void process(String topic, String payload) {
log.debug("处理消息 - 主题: {}, 负载: {}", topic, payload);
public void process(String clientName,String topic, String payload) {
log.debug("处理消息 - 客户端: {}, 主题: {}, 负载: {}", clientName, topic, payload);
try {
// 根据主题选择处理器
MessageHandler handler = findHandler(topic);
if (handler != null) {
handler.handle(topic, payload);
handler.handle(clientName,topic, payload);
} else {
handleDefault(topic, payload);
handleDefault(clientName,topic, payload);
}
} catch (Exception e) {
log.error("处理消息失败", e);
log.error("处理消息失败:{},主题:{},内容:{}",clientName,topic,payload);
}
}
@ -45,8 +45,8 @@ public class MessageProcessor {
* 异步处理消息
*/
@Async
public void processAsync(String topic, String payload) {
process(topic, payload);
public void processAsync(String clientName,String topic, String payload) {
process(clientName,topic, payload);
}
/**
@ -91,8 +91,8 @@ public class MessageProcessor {
/**
* 处理默认消息
*/
private void handleDefault(String topic, String payload) {
log.info("默认处理 - 主题: {}, 消息: {}", topic, payload);
private void handleDefault(String clientName,String topic, String payload) {
log.info("默认处理 - 客户端: {}, 主题: {}, 消息: {}", clientName, topic, payload);
// 可以根据需要存储到数据库或转发到其他系统
// 例如: saveToDatabase(topic, payload);
}
@ -119,6 +119,10 @@ public class MessageProcessor {
return this::handleDeviceRep;
}
if (topic.startsWith(IotInerfaceTopicType.RENKE_TOPIC.getCode())) {
return this::handleRenkeDeviceRep;
}
// // 设备状态主题
// if (topic.contains("/device/") && topic.endsWith("/status")) {
// return this::handleDeviceStatus;
@ -164,14 +168,22 @@ public class MessageProcessor {
@FunctionalInterface
public interface MessageHandler {
void handle(String topic, String payload);
void handle(String clientName,String topic, String payload);
}
/**
* 处理设备监测回执消息
*/
private void handleDeviceRep(String topic, String payload) {
commonService.processMqttData(topic,payload);
private void handleDeviceRep(String clientName,String topic, String payload) {
commonService.processMqttData(clientName,topic,payload);
}
/**
* 处理仁科设备监测回执消息
*/
private void handleRenkeDeviceRep(String clientName,String topic, String payload) {
commonService.processRenkeMqttData(clientName,topic,payload);
}
}

212
src/main/java/com/lanhai/mqtt/service/MqttService.java
View File

@ -2,6 +2,7 @@ package com.lanhai.mqtt.service;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lanhai.mqtt.config.MqttClientManager;
import com.lanhai.mqtt.config.MqttProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@ -20,23 +21,55 @@ import java.util.concurrent.CompletableFuture;
@RequiredArgsConstructor
public class MqttService {
private final MqttClient mqttClient;
private final MqttClientManager mqttClientManager;
private final MqttProperties mqttProperties;
private final ObjectMapper objectMapper;
/**
* 发送消息
* 发送消息（使用默认客户端）
*/
public boolean sendMessage(String topic, String payload) {
return sendMessage(topic, payload, mqttProperties.getQos(), mqttProperties.getRetained());
return sendMessage(null, topic, payload);
}
/**
* 发送消息（自定义QoS）
* 指定客户端发送消息
*/
public boolean sendMessage(String clientName, String topic, String payload) {
MqttClient client = getClient(clientName);
if (client == null) {
log.warn("MQTT 客户端 [{}] 不存在", clientName != null ? clientName : "默认");
return false;
}
MqttProperties.MqttClientConfig config = getClientConfig(clientName);
int qos = config != null && config.getQos() != null ? config.getQos() : 1;
boolean retained = config != null && config.getRetained() != null ? config.getRetained() : false;
return doSendMessage(client, topic, payload, qos, retained);
}
/**
* 发送消息（自定义QoS，使用默认客户端）
*/
public boolean sendMessage(String topic, String payload, int qos, boolean retained) {
return sendMessage(null, topic, payload, qos, retained);
}
/**
* 指定客户端发送消息（自定义QoS）
*/
public boolean sendMessage(String clientName, String topic, String payload, int qos, boolean retained) {
MqttClient client = getClient(clientName);
if (client == null) {
return false;
}
return doSendMessage(client, topic, payload, qos, retained);
}
private boolean doSendMessage(MqttClient client, String topic, String payload, int qos, boolean retained) {
try {
if (!mqttClient.isConnected()) {
if (!client.isConnected()) {
log.warn("MQTT 未连接，消息发送失败: {}", payload);
return false;
}
@ -46,7 +79,7 @@ public class MqttService {
message.setRetained(retained);
message.setId((int) (System.currentTimeMillis() % 100000));
mqttClient.publish(topic, message);
client.publish(topic, message);
log.debug("消息发送成功 - 主题: {}, QoS: {}, 消息: {}", topic, qos,
payload.length() > 100 ? payload.substring(0, 100) + "..." : payload);
return true;
@ -58,12 +91,19 @@ public class MqttService {
}
/**
* 发送 JSON 消息
* 发送 JSON 消息（使用默认客户端）
*/
public boolean sendJsonMessage(String topic, Object data) {
return sendJsonMessage(null, topic, data);
}
/**
* 指定客户端发送 JSON 消息
*/
public boolean sendJsonMessage(String clientName, String topic, Object data) {
try {
String json = objectMapper.writeValueAsString(data);
return sendMessage(topic, json);
return sendMessage(clientName, topic, json);
} catch (JsonProcessingException e) {
log.error("JSON 序列化失败", e);
return false;
@ -71,16 +111,30 @@ public class MqttService {
}
/**
* 异步发送消息
* 异步发送消息（使用默认客户端）
*/
public CompletableFuture<Boolean> sendMessageAsync(String topic, String payload) {
return CompletableFuture.supplyAsync(() -> sendMessage(topic, payload));
return sendMessageAsync(null, topic, payload);
}
/**
* 发送设备控制指令
* 指定客户端异步发送消息
*/
public CompletableFuture<Boolean> sendMessageAsync(String clientName, String topic, String payload) {
return CompletableFuture.supplyAsync(() -> sendMessage(clientName, topic, payload));
}
/**
* 发送设备控制指令（使用默认客户端）
*/
public boolean sendDeviceCommand(String deviceId, String command, Object params) {
return sendDeviceCommand(null, deviceId, command, params);
}
/**
* 指定客户端发送设备控制指令
*/
public boolean sendDeviceCommand(String clientName, String deviceId, String command, Object params) {
String topic = String.format("lanhai/device/%s/command", deviceId);
try {
@ -90,7 +144,7 @@ public class MqttService {
message.put("timestamp", System.currentTimeMillis());
message.put("requestId", UUID.randomUUID().toString());
return sendJsonMessage(topic, message);
return sendJsonMessage(clientName, topic, message);
} catch (Exception e) {
log.error("发送设备指令失败", e);
return false;
@ -98,12 +152,23 @@ public class MqttService {
}
/**
* 订阅主题
* 订阅主题（使用默认客户端）
*/
public boolean subscribe(String topic, int qos) {
return subscribe(null, topic, qos);
}
/**
* 指定客户端订阅主题
*/
public boolean subscribe(String clientName, String topic, int qos) {
MqttClient client = getClient(clientName);
if (client == null) {
return false;
}
try {
mqttClient.subscribe(topic, qos);
log.info("订阅主题: {}", topic);
client.subscribe(topic, qos);
log.info("MQTT 客户端 [{}] 订阅主题: {}", clientName != null ? clientName : "默认", topic);
return true;
} catch (MqttException e) {
log.error("订阅失败", e);
@ -112,12 +177,23 @@ public class MqttService {
}
/**
* 取消订阅
* 取消订阅（使用默认客户端）
*/
public boolean unsubscribe(String topic) {
return unsubscribe(null, topic);
}
/**
* 指定客户端取消订阅
*/
public boolean unsubscribe(String clientName, String topic) {
MqttClient client = getClient(clientName);
if (client == null) {
return false;
}
try {
mqttClient.unsubscribe(topic);
log.info("取消订阅: {}", topic);
client.unsubscribe(topic);
log.info("MQTT 客户端 [{}] 取消订阅: {}", clientName != null ? clientName : "默认", topic);
return true;
} catch (MqttException e) {
log.error("取消订阅失败", e);
@ -126,51 +202,117 @@ public class MqttService {
}
/**
* 检查连接状态
* 检查默认客户端连接状态
*/
public boolean isConnected() {
return mqttClient.isConnected();
return mqttClientManager.isDefaultConnected();
}
/**
* 获取客户端ID
* 检查指定客户端连接状态
*/
public boolean isConnected(String clientName) {
return mqttClientManager.isConnected(clientName);
}
/**
* 获取默认客户端ID
*/
public String getClientId() {
return mqttProperties.getClient().getId();
MqttProperties.MqttClientConfig config = getClientConfig(null);
return config != null && config.getClient() != null ? config.getClient().getId() : null;
}
/**
* 手动重连
* 获取指定客户端ID
*/
public String getClientId(String clientName) {
MqttProperties.MqttClientConfig config = getClientConfig(clientName);
return config != null && config.getClient() != null ? config.getClient().getId() : null;
}
/**
* 手动重连默认客户端
*/
public boolean reconnect() {
return reconnect(null);
}
/**
* 手动重连指定客户端
*/
public boolean reconnect(String clientName) {
MqttClient client = getClient(clientName);
if (client == null) {
return false;
}
try {
if (!mqttClient.isConnected()) {
mqttClient.reconnect();
log.info("手动重连成功");
if (!client.isConnected()) {
client.reconnect();
log.info("MQTT 客户端 [{}] 手动重连成功", clientName != null ? clientName : "默认");
return true;
}
return false;
} catch (MqttException e) {
log.error("手动重连失败", e);
log.error("MQTT 客户端 [{}] 手动重连失败", clientName, e);
return false;
}
}
/**
* 断开连接
* 断开默认客户端连接
*/
public void disconnect() {
try {
if (mqttClient.isConnected()) {
// 发送离线状态
if (mqttProperties.getTopics().getStatus() != null) {
sendMessage(mqttProperties.getTopics().getStatus(), "offline", 1, true);
disconnect(null);
}
mqttClient.disconnect();
log.info("MQTT 连接已断开");
/**
* 断开指定客户端连接
*/
public void disconnect(String clientName) {
MqttClient client = getClient(clientName);
if (client == null) {
return;
}
try {
if (client.isConnected()) {
MqttProperties.MqttClientConfig config = getClientConfig(clientName);
if (config != null && config.getTopics() != null && config.getTopics().getStatus() != null) {
sendMessage(clientName, config.getTopics().getStatus(), "offline", 1, true);
}
client.disconnect();
log.info("MQTT 客户端 [{}] 连接已断开", clientName != null ? clientName : "默认");
}
} catch (MqttException e) {
log.error("断开连接失败", e);
}
}
/**
* 获取客户端
*/
private MqttClient getClient(String clientName) {
if (clientName == null || clientName.isEmpty()) {
return mqttClientManager.getDefaultClient();
}
return mqttClientManager.getClient(clientName);
}
/**
* 获取客户端配置
*/
private MqttProperties.MqttClientConfig getClientConfig(String clientName) {
if (clientName == null || clientName.isEmpty()) {
String defaultClient = mqttProperties.getDefaultClient();
if (defaultClient != null) {
return mqttClientManager.getClientConfig(defaultClient);
}
// 返回第一个配置
if (mqttProperties.getClients() != null && !mqttProperties.getClients().isEmpty()) {
return mqttProperties.getClients().get(0);
}
return null;
}
return mqttClientManager.getClientConfig(clientName);
}
}

2
src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java
View File

@ -15,7 +15,7 @@ import java.util.List;
*/
public interface ISurvIotVirtualDeviceService extends IService<SurvIotVirtualDevice> {
List<String> getAllMqttTopic();
List<String> getAllMqttTopic(String clientName);
void processPayLoad(String topic, byte[] payload);
}

132
src/main/java/com/lanhai/service/Impl/CommonServiceImpl.java
View File

@ -9,8 +9,12 @@ import com.lanhai.constant.PollutionConstants;
import com.lanhai.entity.*;
import com.lanhai.o.iot.lhiot.ResponseCmd;
import com.lanhai.o.iot.pbs.WaterCommonTransVo;
import com.lanhai.o.iot.renke.DataItem;
import com.lanhai.o.iot.renke.RegisterItem;
import com.lanhai.o.iot.renke.RenkeDataRealTimeDetail;
import com.lanhai.service.*;
import com.lanhai.util.LhIotUtil;
import com.lanhai.util.SdrkUtils;
import com.xxl.job.core.context.XxlJobHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
@ -20,8 +24,7 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.*;
/**
* 处理业务的服务
@ -107,14 +110,14 @@ public class CommonServiceImpl implements ICommonService {
}
public void processMqttData(String topic,String mqttMessage) {
public void processMqttData(String clientName,String topic,String mqttMessage) {
//1. 检查主题所属，确定如何执行后续逻辑
List<SurvDeviceDeploy> deployList = deviceDeployService.lambdaQuery()
.eq(SurvDeviceDeploy::getDeviceReverseIotUrl,topic)
.list();
;
log.error("===================主题：{}，设备数量：{}==============================",topic,deployList.size());
XxlJobHelper.log("===================主题：{}，设备数量：{}==============================",topic,deployList.size());
log.error("============客户端：{}=======主题：{}，设备数量：{}==============================",clientName,topic,deployList.size());
XxlJobHelper.log("============客户端：{}=======主题：{}，设备数量：{}==============================",clientName,topic,deployList.size());
if(!deployList.isEmpty()){
SurvDeviceDeploy survDeviceDeploy = deployList.get(0);
@ -126,19 +129,20 @@ public class CommonServiceImpl implements ICommonService {
waterCommonTransVo = LhIotUtil.transData(zhibiaoList,mqttMessage);
break;
}
log.error("----------------------查询到设备：{}，回执:{}----------------------------",survDeviceDeploy.getId()+"_"+survDeviceDeploy.getDeployDes(), JSONUtil.toJsonStr(waterCommonTransVo));
XxlJobHelper.log("----------------------查询到设备：{}，回执:{}----------------------------",survDeviceDeploy.getId()+"_"+survDeviceDeploy.getDeployDes(), JSONUtil.toJsonStr(waterCommonTransVo));
log.error("------------客户端：{}----------查询到设备：{}，回执:{}----------------------------",clientName,survDeviceDeploy.getId()+"_"+survDeviceDeploy.getDeployDes(), JSONUtil.toJsonStr(waterCommonTransVo));
XxlJobHelper.log("-------------客户端：{}---------查询到设备：{}，回执:{}----------------------------",clientName,survDeviceDeploy.getId()+"_"+survDeviceDeploy.getDeployDes(), JSONUtil.toJsonStr(waterCommonTransVo));
if(waterCommonTransVo!=null){
LocalDateTime nowTime =LocalDateTime.now();
List<SurvDeviceDeploy> updateList = new ArrayList<>();
//遍历保存所有租户数据
for (SurvDeviceDeploy deploy : deployList) {
log.error("###########################保存数据中-----设备:{}#####################################",deploy.getDeployCode()+"__"+deploy.getDeployDes());
XxlJobHelper.log("###########################保存数据中-----设备:{}#####################################",deploy.getDeployCode()+"__"+deploy.getDeployDes());
log.error("#############客户端：{}##############保存数据中-----设备:{}#####################################",clientName,deploy.getDeployCode()+"__"+deploy.getDeployDes());
XxlJobHelper.log("###############客户端：{}############保存数据中-----设备:{}#####################################",clientName,deploy.getDeployCode()+"__"+deploy.getDeployDes());
SurvStationInfo survStationInfo = survStationInfoService.getByCode(deploy.getStationCode());
String stationName = survStationInfo!=null?survStationInfo.getStationName():"站点";
if(PollutionConstants.WATER_ORIENT.equals(deploy.getDeployType())){
log.error("^^^^^^^^^^^^^^^^面源数据保存：{}^^^^^^^^^^^^^^^^^",deploy.getDeployCode()+"__"+deploy.getDeployDes());
XxlJobHelper.log("^^^^^^^^^^^^^^^^面源数据保存：{}^^^^^^^^^^^^^^^^^",deploy.getDeployCode()+"__"+deploy.getDeployDes());
log.error("^^^^^^^^^^客户端：{}^^^^^^面源数据保存：{}^^^^^^^^^^^^^^^^^",clientName,deploy.getDeployCode()+"__"+deploy.getDeployDes());
XxlJobHelper.log("^^^^^^^^^客户端：{}^^^^^^^面源数据保存：{}^^^^^^^^^^^^^^^^^",clientName,deploy.getDeployCode()+"__"+deploy.getDeployDes());
SurvTransdataOrientwater oldData = transdataOrientwaterService.getOneByDeviceId(deploy.getId());
if(oldData!=null){
transdataOrientwaterService.removeById(oldData.getId());
@ -151,7 +155,7 @@ public class CommonServiceImpl implements ICommonService {
orientwater.setStationName(stationName);
orientwater.setDeviceName(deploy.getDeployDes());
orientwater.setDeviceId(deploy.getId());
orientwater.setCreatedBy("task");//创建人
orientwater.setCreatedBy("lhmqtt");//创建人
orientwater.setCreateTime(nowTime);//创建时间
orientwater.setStationId(survStationInfo.getId());
orientwater.setDataType(deploy.getDeploySecondaryType());
@ -164,9 +168,11 @@ public class CommonServiceImpl implements ICommonService {
hisOrientWater.setTransDate(nowTime);
hisdataOrientwaterService.save(hisOrientWater);
updateList.add(deploy);
} else if (PollutionConstants.WATER_LIVE.equals(deploy.getDeployType())) {
log.error("^^^^^^^^^^^^^^^^畜禽数据保存：{}^^^^^^^^^^^^^^^^^",deploy.getDeployCode()+"__"+deploy.getDeployDes());
XxlJobHelper.log("^^^^^^^^^^^^^^^^畜禽数据保存：{}^^^^^^^^^^^^^^^^^",deploy.getDeployCode()+"__"+deploy.getDeployDes());
log.error("^^^^^^^^^客户端：{}^^^^^^^畜禽数据保存：{}^^^^^^^^^^^^^^^^^",clientName,deploy.getDeployCode()+"__"+deploy.getDeployDes());
XxlJobHelper.log("^^^^^^^^客户端：{}^^^^^^^^畜禽数据保存：{}^^^^^^^^^^^^^^^^^",clientName,deploy.getDeployCode()+"__"+deploy.getDeployDes());
SurvTransdataLivestockwater oldData = transdataLivestockwaterService.getOneByDeviceId(deploy.getId());
if(oldData!=null){
transdataLivestockwaterService.removeById(oldData.getId());
@ -179,7 +185,7 @@ public class CommonServiceImpl implements ICommonService {
livestockwater.setStationName(stationName);
livestockwater.setDeviceName(deploy.getDeployDes());
livestockwater.setDeviceId(deploy.getId());
livestockwater.setCreatedBy("task");//创建人
livestockwater.setCreatedBy("lhmqtt");//创建人
livestockwater.setCreateTime(nowTime);//创建时间
livestockwater.setDataType(deploy.getDeploySecondaryType());
livestockwater.setStationId(survStationInfo.getId());
@ -191,13 +197,27 @@ public class CommonServiceImpl implements ICommonService {
hisdataLivestockwater.setId(IdUtil.getSnowflakeNextIdStr());
hisdataLivestockwater.setTransDate(nowTime);
hisdataLivestockwaterService.save(hisdataLivestockwater);
updateList.add(deploy);
}
}
if(!updateList.isEmpty()){//更新设备时间
List<SurvDeviceDeploy> upd = new ArrayList<>();
Date now = new Date();
for (SurvDeviceDeploy deviceDeploy : updateList) {
SurvDeviceDeploy ents = new SurvDeviceDeploy();
ents.setId(deviceDeploy.getId());
ents.setLastsyncTime(now);
upd.add(ents);
}
deviceDeployService.updateBatchById(upd);
}
}else{
log.error("主题：{}--消息：{},解析失败，任务中断。",topic,mqttMessage);
XxlJobHelper.log("主题：{}--消息：{},解析失败，任务中断。",topic,mqttMessage);
log.error("客户端：{},主题：{}--消息：{},解析失败，任务中断。",clientName,topic,mqttMessage);
XxlJobHelper.log("客户端：{},主题：{}--消息：{},解析失败，任务中断。",clientName,topic,mqttMessage);
}
}
@ -206,4 +226,82 @@ public class CommonServiceImpl implements ICommonService {
//2. 检查有
}
public void processRenkeMqttData(String clientName,String topic,String mqttMessage) {
//1.数据过滤，根据消息内的设备地址，筛选出对应的设备
RenkeDataRealTimeDetail realTimeDetail = JSONUtil.toBean(mqttMessage,RenkeDataRealTimeDetail.class);
if(realTimeDetail!=null){
if(realTimeDetail.getDeviceAddr()!=null){
String deviceAddr = realTimeDetail.getDeviceAddr().toString();
//2. 匹配设备
List<SurvDeviceDeploy> deployList = deviceDeployService.lambdaQuery()
.in(SurvDeviceDeploy::getProtocolCode,IotConstants.renke_standard)
.eq(SurvDeviceDeploy::getDeployCode,deviceAddr)
.list();
;
log.error("============客户端：{}=======主题：{}，设备号：{}，设备数量：{}==============================",clientName,topic,deviceAddr,deployList.size());
if(!deployList.isEmpty()){
List<SurvDeviceDeploy> updateList = new ArrayList<>();
Map<String, SurvDeviceDeploy> deviceMap = new HashMap<>();
Map<String, List<SurvDeviceDeploy>> allDeviceMap = new HashMap<>();
Map<String,SurvDeviceDeploy> deviceDeployMap = new HashMap<>();
for (SurvDeviceDeploy singleDevice : deployList) {
SurvDeviceDeploy checkDeploy = deviceMap.get(singleDevice.getDeployCode());
Boolean checkResult = Boolean.TRUE;
List<SurvDeviceDeploy> codeDeviceLsit =allDeviceMap.get(singleDevice.getDeployCode());
deviceDeployMap.put(singleDevice.getDeployCode() + "_" + singleDevice.getDeployType(),singleDevice);
if(codeDeviceLsit==null){
codeDeviceLsit = new ArrayList<>();
}
codeDeviceLsit.add(singleDevice);
allDeviceMap.put(singleDevice.getDeployCode(),codeDeviceLsit);
if (checkDeploy != null) {
log.error("发现重复设备,编号:{},已有类型:{},现类型:{}", singleDevice.getDeployCode(), checkDeploy.getDeployType(), singleDevice.getDeployType());
if (PollutionConstants.WATER_QULITY.equals(checkDeploy.getDeployType())) {//如果Map中已有水类型同编号的设备，则不进行替换
checkResult = Boolean.FALSE;
}
}
if (checkResult) {//无异常则进行替换或保存
deviceMap.put(singleDevice.getDeployCode(), singleDevice);
}
}
for (Map.Entry<String, SurvDeviceDeploy> deployEntry : deviceMap.entrySet()) {
SurvDeviceDeploy deploy = deployEntry.getValue();
List<RenkeDataRealTimeDetail> datas = new ArrayList<>();
datas.add(realTimeDetail);
boolean isOk = SdrkUtils.processMqttSurvData(deploy,allDeviceMap,deviceDeployMap,datas);
if(isOk) {
updateList.add(deploy);
}
}
if(!updateList.isEmpty()){//更新设备时间
List<SurvDeviceDeploy> upd = new ArrayList<>();
Date nowTime = new Date();
for(SurvDeviceDeploy deploy:updateList){
SurvDeviceDeploy ents = new SurvDeviceDeploy();
ents.setId(deploy.getId());
ents.setLastsyncTime(nowTime);
upd.add(ents);
}
deviceDeployService.updateBatchById(upd);
}
}
}
}else{
log.error("客户端：{},主题：{}--消息：{},解析失败，任务中断。",clientName,topic,mqttMessage);
}
}
}

21
src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java
View File

@ -44,15 +44,32 @@ public class SurvIotVirtualDeviceServiceImpl extends ServiceImpl<SurvIotVirtualD
private SurvTransdataOrientwaterServiceImpl orientwaterService;
@Autowired
private SurvTransdataLivestockwaterServiceImpl livestockwaterService;
@Autowired
private SurvConfigServiceImpl configService;
@Override
public List<String> getAllMqttTopic() {
public List<String> getAllMqttTopic(String clientName) {
List<String> topics = new ArrayList<>();
//获取所有的mqtt连接配置
List<SurvConfig> configs = configService
.lambdaQuery()
.eq(SurvConfig::getIsDel,0)
.eq(SurvConfig::getConfigValueSe2, clientName)//匹配客户端名称
.eq(SurvConfig::getConfigSubType, IotConstants.CONFIG_TYPE_MQTT)
.list();
List<String> mqttConfigs = new ArrayList<>();
if(!configs.isEmpty()){
mqttConfigs = configs.stream().map(SurvConfig::getId).collect(Collectors.toList());
}
log.error("客户端:{}查询到已配置的mqtt配置:{}个",clientName,mqttConfigs.size());
//已部署设备中配置的主题
List<SurvDeviceDeploy> deploys = deviceDeployService.lambdaQuery()
.eq(SurvDeviceDeploy::getIsDel,0)
.eq(SurvDeviceDeploy::getRunStatus,0)
.eq(SurvDeviceDeploy::getProtocolType, IotProtocolType.MQTT.getCode())
// .eq(SurvDeviceDeploy::getProtocolType, IotProtocolType.MQTT.getCode())
.in(SurvDeviceDeploy::getSurvConfigId,mqttConfigs)
.list();
if(!deploys.isEmpty()){
for (SurvDeviceDeploy deploy : deploys) {

4
src/main/java/com/lanhai/service/Impl/SurvTransdataSoilServiceImpl.java
View File

@ -142,6 +142,7 @@ public class SurvTransdataSoilServiceImpl extends ServiceImpl<SurvTransdataSoilM
BeanUtil.copyProperties(soil,hisdataSoil);
hisdataSoil.setId(IdUtil.getSnowflakeNextIdStr());
hisdataSoil.setCreateTime(nowTime);
hisdataSoil.setTransDate(nowTime);
hisdataSoilService.save(hisdataSoil);
}else{
String dateStr = kDateTime!=null?DateUtil.format(kDateTime,"yyyy-MM-dd HH:mm:ss"):"";
@ -220,7 +221,6 @@ public class SurvTransdataSoilServiceImpl extends ServiceImpl<SurvTransdataSoilM
hisdataAir.setId(IdUtil.getSnowflakeNextIdStr());
hisdataAir.setCreateTime(nowTime);
hisdataAir.setTransDate(nowTime);
hisdataAir.setDataDateTime(nowTime);
hisdataAirService.save(hisdataAir);
}else{
log.error("****** 跳过，数据日期为:{},已存在于空气历史表******",formatAirTime);
@ -310,6 +310,7 @@ public class SurvTransdataSoilServiceImpl extends ServiceImpl<SurvTransdataSoilM
BeanUtil.copyProperties(soil,hisdataSoil);
hisdataSoil.setId(IdUtil.getSnowflakeNextIdStr());
hisdataSoil.setCreateTime(nowTime);
hisdataSoil.setTransDate(nowTime);
hisdataSoilService.save(hisdataSoil);
}else{
String dateStr = kDateTime!=null?DateUtil.format(kDateTime,"yyyy-MM-dd HH:mm:ss"):"";
@ -387,6 +388,7 @@ public class SurvTransdataSoilServiceImpl extends ServiceImpl<SurvTransdataSoilM
BeanUtil.copyProperties(air,hisdataAir);
hisdataAir.setId(IdUtil.getSnowflakeNextIdStr());
hisdataAir.setCreateTime(nowTime);
hisdataAir.setTransDate(nowTime);
hisdataAirService.save(hisdataAir);
}else{
String dateStr = airkDateTime!=null?DateUtil.format(airkDateTime,"yyyy-MM-dd HH:mm:ss"):"";

4
src/main/java/com/lanhai/task/MultithreadTask.java
View File

@ -66,12 +66,12 @@ public class MultithreadTask {
deviceType.add(PollutionConstants.AIR_SURV);
deviceType.add(PollutionConstants.SOIL_SURV);
deviceType.add(PollutionConstants.WATER_QULITY);
deviceType.add(PollutionConstants.FLOWMETER);
// deviceType.add(PollutionConstants.FLOWMETER); 改用mqtt
List<String> protocolList = new ArrayList<String>(){
{add(IotConstants.xph_v2);
add(IotConstants.tuya_standard);
add(IotConstants.renke_standard);
// add(IotConstants.renke_standard); 改用mqtt
}
};
List<SurvDeviceDeploy> deploys = deviceDeployService.getDeviceByType(protocolList,deviceType);

54
src/main/java/com/lanhai/util/DateUtils.java
View File

@ -801,4 +801,58 @@ public class DateUtils extends PropertyEditorSupport {
public static long getDaysDiff(Date startDate, Date endDate) {
return getDateDiff(startDate, endDate, TimeUnit.DAYS);
}
/**
* 计算目标时间相对于基准时间的秒数差，并标明正负
*
* @param baseTime 基准时间（相当于"当前时间"）
* @param targetTime 目标时间
* @return 秒数，正数表示 targetTime 在 baseTime 之后（未来），
* 负数表示 targetTime 在 baseTime 之前（过去），
* 0 表示两者相同（或差值小于1秒）
* @throws IllegalArgumentException 如果任一参数为 null
*/
public static long getSecondsBetween(Date baseTime, Date targetTime) {
if (baseTime == null || targetTime == null) {
throw new IllegalArgumentException("baseTime and targetTime cannot be null");
}
long diffMillis = targetTime.getTime() - baseTime.getTime();
return TimeUnit.MILLISECONDS.toSeconds(diffMillis);
}
/**
* 精确版本（返回 double，保留毫秒级精度）
*
* @param baseTime 基准时间
* @param targetTime 目标时间
* @return 秒数（double），正数未来，负数过去
*/
public static double getSecondsBetweenPrecise(Date baseTime, Date targetTime) {
if (baseTime == null || targetTime == null) {
throw new IllegalArgumentException("baseTime and targetTime cannot be null");
}
long diffMillis = targetTime.getTime() - baseTime.getTime();
return diffMillis / 1000.0;
}
/**
* 获取带正负号描述的字符串（便于日志输出）
*
* @param baseTime 基准时间
* @param targetTime 目标时间
* @return 描述字符串，如 "+30秒"、"-15秒"、"0秒"
*/
public static String getSecondsDesc(Date baseTime, Date targetTime) {
long seconds = getSecondsBetween(baseTime, targetTime);
if (seconds > 0) {
return "+" + seconds + "秒";
} else if (seconds < 0) {
return seconds + "秒"; // 已经是负号
} else {
return "0秒";
}
}
}

40
src/main/java/com/lanhai/util/SdrkUtils.java
View File

@ -20,6 +20,7 @@ import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
@ -68,6 +69,39 @@ public class SdrkUtils {
return b;
}
public static boolean processMqttSurvData(SurvDeviceDeploy deploy, Map<String, List<SurvDeviceDeploy>> allDeviceMap, Map<String, SurvDeviceDeploy> deviceDeployMap, List<RenkeDataRealTimeDetail> datas) {
List<SurvDeviceDeploy> allDeployList = allDeviceMap.get(deploy.getDeployCode());
deploy.setDeviceList(allDeployList);
log.error("========正在执行仁科mqtt=======，设备编号:{},类型:{}==========", deploy.getDeployCode(), deploy.getDeployType());
boolean b = false;
if(datas!=null){
if(!datas.isEmpty()){
Date now = new Date();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//此处至少有一条数据，所以直接处理
for (SurvDeviceDeploy survDeviceDeploy : deploy.getDeviceList()) {
//校验设备上次数据时间，超过3分钟才入库
if(survDeviceDeploy.getLastsyncTime() != null){
long seconds = DateUtils.getSecondsBetween(now,survDeviceDeploy.getLastsyncTime());
if(seconds>-180){//时间间隔小于3分钟，跳过
log.error("=======跳过仁科mqtt数据，设备号：{}-{},上次数据时间：{}==========",deploy.getId(),deploy.getDeployCode(),sdf.format(survDeviceDeploy.getLastsyncTime()));
continue;
}
}
survDeviceDeploy.setScEquZhibiaoList(scEquZhibiaoService.getListByEquid(survDeviceDeploy.getId()));
saveData(survDeviceDeploy,deviceDeployMap,datas);
b = true;
}
}
}
return b;
}
public static void saveData(SurvDeviceDeploy deploy, Map<String, SurvDeviceDeploy> deviceDeployMap, List<RenkeDataRealTimeDetail> datas) {
if(datas!=null && !datas.isEmpty()){
//复用新普惠构造方式，传入参数
@ -96,8 +130,10 @@ public class SdrkUtils {
}
if(dataMap.size()>1){//不值包含mn
if(DeviceDeployEnum.SURV_SOIL.getType().equals(deploy.getDeployType())){//土壤监测设备
if(DeviceDeployEnum.SURV_SOIL.getType().equals(deploy.getDeployType()) || DeviceDeployEnum.SURV_AIR.getType().equals(deploy.getDeployType())){//土壤/空气监测设备
survTransdataSoilService.saveBaowen(dataMap, deploy,deviceDeployMap);
} else if (DeviceDeployEnum.WATER_QULITY.getType().equals(deploy.getDeployType())) {//水监测设备
survTransdataSoilService.saveWaterBaowen(dataMap, deploy,deviceDeployMap);
} else if(DeviceDeployEnum.FLOWMETER.getType().equals(deploy.getDeployType())){
flowmeterService.saveBaowen(dataMap, deploy,deviceDeployMap);
}
@ -243,4 +279,6 @@ public class SdrkUtils {
return null;
}
}

50
src/main/resources/application-local.yml
View File

@ -7,7 +7,7 @@ server:
spring:
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://172.27.17.3:13306/fx_nsp?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
url: jdbc:mysql://172.27.17.13:13306/fx_nsp?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
username: user_fx
password: user_fx
type: com.alibaba.druid.pool.DruidDataSource
@ -77,13 +77,47 @@ xxl:
logpath: /logs/xxl-job/jobhandler
logretentiondays: 30
mqtt:
host: tcp://mqtt.ilhzn.cn:10007
userName: lhzn.mqtt
passWord: lhzn.2025
enabled: true
# 指定默认客户端名称（可选）
default-client: broker1
# 多客户端配置
clients:
- name: lhiotMqtt
enabled: true
broker:
url: "tcp://mqtt.ilhzn.cn:10007"
client:
id: fxgather-client-${random.uuid}
clean-session: true
connection:
timeout: 10
keep-alive: 30
automatic-reconnect: true
auth:
username: "lhzn.mqtt"
password: "lhzn.2025"
topics:
status: "device/status"
qos: 1
clientId: fxgather-client-${random.uuid} #ClientId_local必须唯一。
timeout: 10 # 超时时间
keepalive: 30 # 保持连接时间
clearSession: false # 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息)
retained: false
- name: renkeMqtt
enabled: true
broker:
url: "tcp://rtd.mqtt.0531yun.com:1883"
client:
id: fxgather-client-${random.uuid}
clean-session: true
connection:
timeout: 10
keep-alive: 30
automatic-reconnect: true
auth:
username: "h250830lhzn"
password: "h250830lhzn"
topics:
status: "device/status2"
qos: 1
retained: false

46
src/main/resources/application-prod.yml
View File

@ -79,26 +79,46 @@ xxl:
logretentiondays: 30
# 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息)
mqtt:
enabled: true
# 指定默认客户端名称（可选）
default-client: broker1
# 多客户端配置
clients:
- name: lhiotMqtt
enabled: true
broker:
url: tcp://mqtt.ilhzn.cn:10007
# 多个broker配置: tcp://broker1:1883,tcp://broker2:1883
url: "tcp://mqtt.ilhzn.cn:10007"
client:
id: lanhai-mqtt-client-${random.uuid}
clean-session: false
id: fxgather-client-${random.uuid}
clean-session: true
connection:
timeout: 30
keep-alive: 60
max-reconnect-delay: 128000 # 最大重连延迟(ms)
timeout: 10
keep-alive: 30
automatic-reconnect: true
auth:
username: lhzn.mqtt
password: lhzn.2025
username: "lhzn.mqtt"
password: "lhzn.2025"
topics:
send: "lanhai/mqtt/send"
receive: "lanhai/mqtt/receive"
status: "lanhai/mqtt/status"
device: "lanhai/device/+/command"
status: "device/status"
qos: 1
retained: false
- name: renkeMqtt
enabled: true
broker:
url: "tcp://rtd.mqtt.0531yun.com:1883"
client:
id: fxgather-client-${random.uuid}
clean-session: true
connection:
timeout: 10
keep-alive: 30
automatic-reconnect: true
auth:
username: "h250830lhzn"
password: "h250830lhzn"
topics:
status: "device/status2"
qos: 1
retained: false

10
src/test/java/com/lanhai/test/Test2.java
View File

@ -35,12 +35,10 @@ public class Test2 {
// survTransdataSoilService.saveBaowen(map);
// ip:110.179.80.65,port:51077
String hostAddress = "110.179.80.65";//ip:117.132.191.89
String port = String.valueOf(51077);//port:8515
System.out.println("ip:"+hostAddress+",port:"+port);
ISurvDeviceDeployService deviceDeployService = SpringContextUtil.getBean(ISurvDeviceDeployService.class);
SurvDeviceDeploy deviceDeploy = deviceDeployService.getOneByIpPort(hostAddress,port);
System.out.println(deviceDeploy);
long now = -400;
long f= -300;
System.out.println(f < now);
}
}

10
src/test/java/com/lanhai/test/testway.java Normal file
View File

@ -0,0 +1,10 @@
package com.lanhai.test;
public class testway {
public static void main(String[] args) {
long now = -300;
long f= -400;
System.out.println(f < now);
}
}