From 340abe8a187fc53bbe77859fc6f056cf7f5cb4be Mon Sep 17 00:00:00 2001 From: zhangyue <82248909@qq.com> Date: Wed, 29 Jul 2026 16:28:09 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=A4=9AMqtt=E5=AE=9E?= =?UTF-8?q?=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/lanhai/constant/IotConstants.java | 6 + .../java/com/lanhai/entity/SurvConfig.java | 6 + .../lanhai/enums/IotInerfaceTopicType.java | 1 + .../lanhai/mqtt/config/MqttClientManager.java | 229 ++++++++++++++++++ .../com/lanhai/mqtt/config/MqttConfig.java | 135 +---------- .../lanhai/mqtt/config/MqttProperties.java | 58 ++++- .../mqtt/handler/MqttCallbackHandler.java | 38 +-- .../mqtt/listener/MqttReconnectListener.java | 170 +++++++------ .../lanhai/mqtt/service/MessageProcessor.java | 36 ++- .../com/lanhai/mqtt/service/MqttService.java | 210 +++++++++++++--- .../service/ISurvIotVirtualDeviceService.java | 2 +- .../service/Impl/CommonServiceImpl.java | 132 ++++++++-- .../Impl/SurvIotVirtualDeviceServiceImpl.java | 21 +- .../Impl/SurvTransdataSoilServiceImpl.java | 4 +- .../java/com/lanhai/task/MultithreadTask.java | 4 +- src/main/java/com/lanhai/util/DateUtils.java | 54 +++++ src/main/java/com/lanhai/util/SdrkUtils.java | 42 +++- src/main/resources/application-local.yml | 52 +++- src/main/resources/application-prod.yml | 62 +++-- src/test/java/com/lanhai/test/Test2.java | 10 +- src/test/java/com/lanhai/test/testway.java | 10 + 21 files changed, 954 insertions(+), 328 deletions(-) create mode 100644 src/main/java/com/lanhai/mqtt/config/MqttClientManager.java create mode 100644 src/test/java/com/lanhai/test/testway.java diff --git a/src/main/java/com/lanhai/constant/IotConstants.java b/src/main/java/com/lanhai/constant/IotConstants.java index 8bc0b5a..905cc52 100644 --- a/src/main/java/com/lanhai/constant/IotConstants.java +++ b/src/main/java/com/lanhai/constant/IotConstants.java @@ -221,4 +221,10 @@ public interface IotConstants { */ String lhviot_standard = "lhviot_standard"; + + /** + * mqtt连接协议 + */ + String CONFIG_TYPE_MQTT = "mqtt"; + } diff --git a/src/main/java/com/lanhai/entity/SurvConfig.java b/src/main/java/com/lanhai/entity/SurvConfig.java index 51c5b9d..2feafdd 100644 --- a/src/main/java/com/lanhai/entity/SurvConfig.java +++ b/src/main/java/com/lanhai/entity/SurvConfig.java @@ -88,6 +88,12 @@ public class SurvConfig extends Model { @TableField("CONFIG_TYPE") private String configType; + /** + * 系统内置 + */ + @TableField("CONFIG_SUB_TYPE") + private String configSubType; + /** * 备注 */ diff --git a/src/main/java/com/lanhai/enums/IotInerfaceTopicType.java b/src/main/java/com/lanhai/enums/IotInerfaceTopicType.java index ecf7ab9..0b91ab0 100644 --- a/src/main/java/com/lanhai/enums/IotInerfaceTopicType.java +++ b/src/main/java/com/lanhai/enums/IotInerfaceTopicType.java @@ -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","其他") ; diff --git a/src/main/java/com/lanhai/mqtt/config/MqttClientManager.java b/src/main/java/com/lanhai/mqtt/config/MqttClientManager.java new file mode 100644 index 0000000..cd80cb4 --- /dev/null +++ b/src/main/java/com/lanhai/mqtt/config/MqttClientManager.java @@ -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 clientMap = new ConcurrentHashMap<>(); + + /** + * 存储每个客户端对应的回调处理器 + */ + private final Map 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 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 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 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 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(); + } +} \ No newline at end of file diff --git a/src/main/java/com/lanhai/mqtt/config/MqttConfig.java b/src/main/java/com/lanhai/mqtt/config/MqttConfig.java index 6b515cc..d585fb6 100644 --- a/src/main/java/com/lanhai/mqtt/config/MqttConfig.java +++ b/src/main/java/com/lanhai/mqtt/config/MqttConfig.java @@ -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 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(); } } \ No newline at end of file diff --git a/src/main/java/com/lanhai/mqtt/config/MqttProperties.java b/src/main/java/com/lanhai/mqtt/config/MqttProperties.java index 3eb3c0b..96768b2 100644 --- a/src/main/java/com/lanhai/mqtt/config/MqttProperties.java +++ b/src/main/java/com/lanhai/mqtt/config/MqttProperties.java @@ -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 Broker broker = new Broker(); - private Client client = new Client(); - private Connection connection = new Connection(); - private Auth auth = new Auth(); - private Topics topics = new Topics(); - private Integer qos; - private Boolean retained; + /** + * 默认客户端（向后兼容单连接配置） + */ + private String defaultClient; + /** + * 多客户端配置列表 + */ + private List 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(); + /** + * QoS + */ + private Integer qos = 1; + /** + * Retained + */ + private Boolean retained = false; + } @Data public static class Broker { diff --git a/src/main/java/com/lanhai/mqtt/handler/MqttCallbackHandler.java b/src/main/java/com/lanhai/mqtt/handler/MqttCallbackHandler.java index 5725119..ab7d312 100644 --- a/src/main/java/com/lanhai/mqtt/handler/MqttCallbackHandler.java +++ b/src/main/java/com/lanhai/mqtt/handler/MqttCallbackHandler.java @@ -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); } diff --git a/src/main/java/com/lanhai/mqtt/listener/MqttReconnectListener.java b/src/main/java/com/lanhai/mqtt/listener/MqttReconnectListener.java index 46e0bbf..08be34f 100644 --- a/src/main/java/com/lanhai/mqtt/listener/MqttReconnectListener.java +++ b/src/main/java/com/lanhai/mqtt/listener/MqttReconnectListener.java @@ -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 reconnectingMap = new ConcurrentHashMap<>(); + private final Map 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 clientNames = mqttClientManager.getAllClientNames(); + for (String name : clientNames) { + log.info("MQTT 客户端 [{}] 连接状态: {}", name, mqttClientManager.isConnected(name)); + } } /** - * 定时检查连接状态并重连 + * 定时检查所有客户端连接状态并重连 */ @Scheduled(fixedDelay = 10000) // 每10秒检查一次 public void checkAndReconnect() { - // 如果已经连接，重置重连计数 - if (mqttClient.isConnected()) { - if (reconnectCount > 0) { - log.info("MQTT 连接已恢复，重置重连计数器"); - reconnectCount = 0; - isReconnecting.set(false); + List clientNames = mqttClientManager.getAllClientNames(); + for (String clientName : clientNames) { + MqttClient client = mqttClientManager.getClient(clientName); + if (client == null) continue; + + // 如果已经连接，重置重连计数 + 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)); + } + continue; } - return; - } - // 如果正在重连，跳过 - if (isReconnecting.get()) { - return; - } + // 如果正在重连，跳过 + AtomicBoolean isReconnecting = reconnectingMap.computeIfAbsent(clientName, k -> new AtomicBoolean(false)); + if (isReconnecting.get()) { + continue; + } - // 检查重连次数 - if (reconnectCount >= MAX_RECONNECT_COUNT) { - log.error("MQTT 重连失败次数已达上限 ({})，停止重连", MAX_RECONNECT_COUNT); - return; - } + // 检查重连次数 + int reconnectCount = reconnectCountMap.getOrDefault(clientName, 0); + if (reconnectCount >= MAX_RECONNECT_COUNT) { + 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 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); } } \ No newline at end of file diff --git a/src/main/java/com/lanhai/mqtt/service/MessageProcessor.java b/src/main/java/com/lanhai/mqtt/service/MessageProcessor.java index 1d3c19d..20904e6 100644 --- a/src/main/java/com/lanhai/mqtt/service/MessageProcessor.java +++ b/src/main/java/com/lanhai/mqtt/service/MessageProcessor.java @@ -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); } } \ No newline at end of file diff --git a/src/main/java/com/lanhai/mqtt/service/MqttService.java b/src/main/java/com/lanhai/mqtt/service/MqttService.java index ee8ad34..26a3464 100644 --- a/src/main/java/com/lanhai/mqtt/service/MqttService.java +++ b/src/main/java/com/lanhai/mqtt/service/MqttService.java @@ -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 sendMessageAsync(String topic, String payload) { - return CompletableFuture.supplyAsync(() -> sendMessage(topic, payload)); + return sendMessageAsync(null, topic, payload); } /** - * 发送设备控制指令 + * 指定客户端异步发送消息 + */ + public CompletableFuture 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() { + disconnect(null); + } + + /** + * 断开指定客户端连接 + */ + public void disconnect(String clientName) { + MqttClient client = getClient(clientName); + if (client == null) { + return; + } try { - if (mqttClient.isConnected()) { - // 发送离线状态 - if (mqttProperties.getTopics().getStatus() != null) { - sendMessage(mqttProperties.getTopics().getStatus(), "offline", 1, true); + 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); } - mqttClient.disconnect(); - log.info("MQTT 连接已断开"); + 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); + } } \ No newline at end of file diff --git a/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java b/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java index ed9e288..bbba313 100644 --- a/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java +++ b/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java @@ -15,7 +15,7 @@ import java.util.List; */ public interface ISurvIotVirtualDeviceService extends IService { - List getAllMqttTopic(); + List getAllMqttTopic(String clientName); void processPayLoad(String topic, byte[] payload); } diff --git a/src/main/java/com/lanhai/service/Impl/CommonServiceImpl.java b/src/main/java/com/lanhai/service/Impl/CommonServiceImpl.java index 79262d8..0c17d64 100644 --- a/src/main/java/com/lanhai/service/Impl/CommonServiceImpl.java +++ b/src/main/java/com/lanhai/service/Impl/CommonServiceImpl.java @@ -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 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 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 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 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 updateList = new ArrayList<>(); + Map deviceMap = new HashMap<>(); + Map> allDeviceMap = new HashMap<>(); + Map deviceDeployMap = new HashMap<>(); + + for (SurvDeviceDeploy singleDevice : deployList) { + SurvDeviceDeploy checkDeploy = deviceMap.get(singleDevice.getDeployCode()); + Boolean checkResult = Boolean.TRUE; + List 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 deployEntry : deviceMap.entrySet()) { + SurvDeviceDeploy deploy = deployEntry.getValue(); + List datas = new ArrayList<>(); + datas.add(realTimeDetail); + boolean isOk = SdrkUtils.processMqttSurvData(deploy,allDeviceMap,deviceDeployMap,datas); + if(isOk) { + updateList.add(deploy); + } + } + + + if(!updateList.isEmpty()){//更新设备时间 + List 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); + } + + + + } } diff --git a/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java index 83fccbd..ad2227e 100644 --- a/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java +++ b/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java @@ -44,15 +44,32 @@ public class SurvIotVirtualDeviceServiceImpl extends ServiceImpl getAllMqttTopic() { + public List getAllMqttTopic(String clientName) { List topics = new ArrayList<>(); + //获取所有的mqtt连接配置 + List configs = configService + .lambdaQuery() + .eq(SurvConfig::getIsDel,0) + .eq(SurvConfig::getConfigValueSe2, clientName)//匹配客户端名称 + .eq(SurvConfig::getConfigSubType, IotConstants.CONFIG_TYPE_MQTT) + .list(); + List mqttConfigs = new ArrayList<>(); + if(!configs.isEmpty()){ + mqttConfigs = configs.stream().map(SurvConfig::getId).collect(Collectors.toList()); + } + log.error("客户端:{}查询到已配置的mqtt配置:{}个",clientName,mqttConfigs.size()); + //已部署设备中配置的主题 List 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) { diff --git a/src/main/java/com/lanhai/service/Impl/SurvTransdataSoilServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvTransdataSoilServiceImpl.java index 8ba4282..9582305 100644 --- a/src/main/java/com/lanhai/service/Impl/SurvTransdataSoilServiceImpl.java +++ b/src/main/java/com/lanhai/service/Impl/SurvTransdataSoilServiceImpl.java @@ -142,6 +142,7 @@ public class SurvTransdataSoilServiceImpl extends ServiceImpl protocolList = new ArrayList(){ {add(IotConstants.xph_v2); add(IotConstants.tuya_standard); - add(IotConstants.renke_standard); +// add(IotConstants.renke_standard); 改用mqtt } }; List deploys = deviceDeployService.getDeviceByType(protocolList,deviceType); diff --git a/src/main/java/com/lanhai/util/DateUtils.java b/src/main/java/com/lanhai/util/DateUtils.java index ec0c13a..aa033d4 100644 --- a/src/main/java/com/lanhai/util/DateUtils.java +++ b/src/main/java/com/lanhai/util/DateUtils.java @@ -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秒"; + } + } } \ No newline at end of file diff --git a/src/main/java/com/lanhai/util/SdrkUtils.java b/src/main/java/com/lanhai/util/SdrkUtils.java index 401c167..59dff0b 100644 --- a/src/main/java/com/lanhai/util/SdrkUtils.java +++ b/src/main/java/com/lanhai/util/SdrkUtils.java @@ -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> allDeviceMap, Map deviceDeployMap, List datas) { + List 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 deviceDeployMap, List datas) { if(datas!=null && !datas.isEmpty()){ //复用新普惠构造方式，传入参数 @@ -96,9 +130,11 @@ 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.FLOWMETER.getType().equals(deploy.getDeployType())){ + } 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; } + + } diff --git a/src/main/resources/application-local.yml b/src/main/resources/application-local.yml index 1874e3a..1dcffd7 100644 --- a/src/main/resources/application-local.yml +++ b/src/main/resources/application-local.yml @@ -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 - qos: 1 - clientId: fxgather-client-${random.uuid} #ClientId_local必须唯一。 - timeout: 10 # 超时时间 - keepalive: 30 # 保持连接时间 - clearSession: false # 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息) + 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 + 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 diff --git a/src/main/resources/application-prod.yml b/src/main/resources/application-prod.yml index 4ff4d89..2f6dbde 100644 --- a/src/main/resources/application-prod.yml +++ b/src/main/resources/application-prod.yml @@ -80,26 +80,46 @@ xxl: # 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息) mqtt: enabled: true - broker: - url: tcp://mqtt.ilhzn.cn:10007 - # 多个broker配置: tcp://broker1:1883,tcp://broker2:1883 - client: - id: lanhai-mqtt-client-${random.uuid} - clean-session: false - connection: - timeout: 30 - keep-alive: 60 - max-reconnect-delay: 128000 # 最大重连延迟(ms) - automatic-reconnect: true - auth: - username: lhzn.mqtt - password: lhzn.2025 - topics: - send: "lanhai/mqtt/send" - receive: "lanhai/mqtt/receive" - status: "lanhai/mqtt/status" - device: "lanhai/device/+/command" - qos: 1 - retained: false + # 指定默认客户端名称（可选） + 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 + 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 diff --git a/src/test/java/com/lanhai/test/Test2.java b/src/test/java/com/lanhai/test/Test2.java index e2c8bd6..50e8650 100644 --- a/src/test/java/com/lanhai/test/Test2.java +++ b/src/test/java/com/lanhai/test/Test2.java @@ -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); + } } diff --git a/src/test/java/com/lanhai/test/testway.java b/src/test/java/com/lanhai/test/testway.java new file mode 100644 index 0000000..790bdb0 --- /dev/null +++ b/src/test/java/com/lanhai/test/testway.java @@ -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); + } +}