修改MQTT请求

src/main/java/cn/com/GogoApplication.java

@@ -3,8 +3,10 @@ package cn.com;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication
+@EnableScheduling
 @MapperScan({"cn.com.v2.mapper", "cn.com.mes.mapper", "cn.com.energy.mapper", "cn.com.oa.mapper"})
 public class GogoApplication {
 

src/main/java/cn/com/mes/controller/BigViewWmsController.java

@@ -1,5 +1,6 @@
 package cn.com.mes.controller;
 
+import cn.com.mes.server.HumitureServer;
 import cn.com.mes.service.*;
 import cn.com.v2.common.base.BaseController;
 import cn.com.v2.common.domain.AjaxResult;
@@ -15,7 +16,7 @@ import java.util.Map;
 
 /**
  * WMS大屏统计接口控制器
- * 
+ *
  * @author system
  * @date 2024-12-19
  */
@@ -23,22 +24,22 @@ import java.util.Map;
 @RestController
 @RequestMapping("/api/wms")
 public class BigViewWmsController extends BaseController {
-    
-    @Autowired 
+
+    @Autowired
     private IWmItemRecptService wmItemRecptService;
-    @Autowired 
+    @Autowired
     private IWmProductRecptService wmProductRecptService;
-    @Autowired 
+    @Autowired
     private IWmProductSalseService wmProductSalseService;
-    @Autowired 
+    @Autowired
     private IWmIssueService wmIssueService;
-    @Autowired 
-    private IWmRtVendorService wmRtVendorService;
-    @Autowired 
+    @Autowired
+    private HumitureServer humitureServer;
+    @Autowired
     private IWmsCommonService wmsCommonService;
-    @Autowired 
+    @Autowired
     private IWmStorageAreaService wmStorageAreaService;
-    
+
     /**
      * 销售出库总数
      */
@@ -185,11 +186,11 @@ public class BigViewWmsController extends BaseController {
     @GetMapping("/humiture")
     public AjaxResult getHumiture() {
         try {
-            Map<String, Object> result = wmsCommonService.getHumiture();
+            Map<String, Object> result = humitureServer.getHumiture();
             return success().put("data", result);
         } catch (Exception e) {
             return error("获取温度湿度数据失败：" + e.getMessage());
         }
     }
 
-}
+}

src/main/java/cn/com/mes/server/HumitureServer.java

@@ -0,0 +1,145 @@
+package cn.com.mes.server;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Service
+public class HumitureServer {
+    private static final String BROKER = "tcp://222.243.138.146:1883";
+    private static final String TOPIC = "yzkj/zkshi/esp32/DH11";
+    private static final String USERNAME = "ygtx";
+    private static final String PASSWORD = "T!8dFz$kW3#pRgM2";
+
+    private MqttClient client;
+    private final Map<String, Object> latestData = new ConcurrentHashMap<>();
+
+    /**
+     * 程序启动时自动执行，建立 MQTT 长连接
+     */
+    @PostConstruct
+    public void init() {
+        connectAndSubscribe();
+    }
+
+    /**
+     * 建立连接并订阅主题
+     */
+    private void connectAndSubscribe() {
+        try {
+            String clientId = "java_client_" + System.currentTimeMillis();
+            MqttConnectOptions options = new MqttConnectOptions();
+            options.setUserName(USERNAME);
+            options.setPassword(PASSWORD.toCharArray());
+            options.setConnectionTimeout(10);
+            options.setKeepAliveInterval(60);
+            options.setCleanSession(true);
+
+            client = new MqttClient(BROKER, clientId, new MemoryPersistence());
+
+            client.setCallback(new MqttCallback() {
+                @Override
+                public void connectionLost(Throwable cause) {
+                    System.err.println("[MQTT] 连接丢失：" + cause.getMessage());
+                    reconnect();
+                }
+
+                @Override
+                public void messageArrived(String topic, MqttMessage message) {
+                    try {
+                        String payload = new String(message.getPayload());
+                        ObjectMapper mapper = new ObjectMapper();
+                        Map<String, Object> data = mapper.readValue(payload, Map.class);
+                        data.put("statisticsTime", new Date());
+                        latestData.clear();
+                        latestData.putAll(data);
+                        System.out.println("[MQTT] 收到温湿度数据：" + data);
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
+                }
+
+                @Override
+                public void deliveryComplete(IMqttDeliveryToken token) {}
+            });
+
+            client.connect(options);
+            client.subscribe(TOPIC, 0);
+            System.out.println("[MQTT] 已连接并订阅：" + TOPIC);
+
+        } catch (Exception e) {
+            System.err.println("[MQTT] 连接失败：" + e.getMessage());
+            scheduleReconnect();
+        }
+    }
+
+    /**
+     * 自动重连逻辑（延迟5秒）
+     */
+    private void reconnect() {
+        try {
+            if (client != null && client.isConnected()) return;
+            Thread.sleep(5000);
+            System.out.println("[MQTT] 尝试重连...");
+            connectAndSubscribe();
+        } catch (Exception e) {
+            e.printStackTrace();
+            scheduleReconnect();
+        }
+    }
+
+    /**
+     * 当连接失败时，延迟重新连接
+     */
+    private void scheduleReconnect() {
+        new Thread(() -> {
+            try {
+                Thread.sleep(10000);
+                reconnect();
+            } catch (InterruptedException ignored) {}
+        }).start();
+    }
+
+    /**
+     * 对外提供获取最新温湿度的方法
+     */
+    public Map<String, Object> getHumiture() {
+        if (latestData.isEmpty()) {
+            return new HashMap<>();
+        }
+        return new HashMap<>(latestData);
+    }
+
+    /**
+     * 定时任务示例：每5秒打印一次最新数据
+     */
+//    @Scheduled(fixedRate = 5000)
+//    public void printLatestHumiture() {
+//        System.out.println("[定时任务] 最新温湿度数据：" + getHumiture());
+//    }
+
+    /**
+     * 程序关闭时断开连接
+     */
+    @PreDestroy
+    public void shutdown() {
+        try {
+            if (client != null && client.isConnected()) {
+                client.disconnect();
+                client.close();
+                System.out.println("[MQTT] 已断开连接。");
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/cn/com/mes/service/IWmsCommonService.java

@@ -5,8 +5,6 @@ import java.util.Map;
 
 public interface IWmsCommonService {
 
-    Map<String, Object> getHumiture();
-
     List<Map<String, Object>> getDailyStockInOutData();
 
     List<Map<String, Object>> getOverdueOrderStatistics();

src/main/java/cn/com/mes/service/impl/WmsCommonServiceImpl.java

@@ -43,52 +43,6 @@ public class WmsCommonServiceImpl implements IWmsCommonService {
     @Autowired
     private WmMaterialStockMapper wmMaterialStockMapper;
 
-    @Override
-    public Map<String, Object> getHumiture() {
-        try {
-            String broker = "tcp://222.243.138.146:1883";
-            String topic = "yzkj/zkshi/esp32/DH11";
-            String clientId = "java_client_" + System.currentTimeMillis();
-            String username = "ygtx";
-            String password = "T!8dFz$kW3#pRgM2";
-
-            MqttConnectOptions options = new MqttConnectOptions();
-            options.setUserName(username);
-            options.setPassword(password.toCharArray());
-            options.setConnectionTimeout(10);
-            options.setKeepAliveInterval(60);
-            options.setCleanSession(true);
-
-            MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
-
-            CompletableFuture<Map<String, Object>> future = new CompletableFuture<>();
-            client.setCallback(new MqttCallback() {
-                @Override public void connectionLost(Throwable cause) { future.completeExceptionally(cause); }
-                @Override public void messageArrived(String t, MqttMessage msg) {
-                    try {
-                        String payload = new String(msg.getPayload());
-                        ObjectMapper mapper = new ObjectMapper();
-                        Map<String, Object> data = mapper.readValue(payload, Map.class);
-                        data.put("statisticsTime", DateUtil.now());
-                        future.complete(data);
-                    } catch (Exception e) { future.completeExceptionally(e); }
-                }
-                @Override public void deliveryComplete(IMqttDeliveryToken token) {}
-            });
-
-            client.connect(options);
-            client.subscribe(topic, 0);
-
-            Map<String, Object> result = future.get(5, TimeUnit.SECONDS);
-            client.disconnect();
-            client.close();
-            client.disconnectForcibly();
-            return result;
-        } catch (Exception e) {
-            throw new RuntimeException("获取温湿度失败：" + e.getMessage());
-        }
-    }
-
     @Override
     public List<Map<String, Object>> getOverdueOrderStatistics() {
         List<Map<String, Object>> itemReceiptData = wmItemRecptMapper.getOverdueItemReceiptStatistics();
@@ -172,7 +126,7 @@ public class WmsCommonServiceImpl implements IWmsCommonService {
 
         List<Map<String, Object>> result = new ArrayList<>();
         for (Map<String, Object> dayData : dateMap.values()) {
-            int totalOrders = 
+            int totalOrders =
                 ((Number) dayData.get("itemReceiptTotal")).intValue() +
                 ((Number) dayData.get("productReceiptTotal")).intValue() +
                 ((Number) dayData.get("salesTotal")).intValue() +
@@ -180,7 +134,7 @@ public class WmsCommonServiceImpl implements IWmsCommonService {
                 ((Number) dayData.get("returnTotal")).intValue();
             dayData.put("totalOrders", totalOrders);
 
-            int totalOverdue = 
+            int totalOverdue =
                 ((Number) dayData.get("itemReceiptOverdue")).intValue() +
                 ((Number) dayData.get("productReceiptOverdue")).intValue() +
                 ((Number) dayData.get("salesOverdue")).intValue() +

src/main/java/cn/com/oa/common/constant/DataLoaderRunner.java

@@ -24,13 +24,17 @@ public class DataLoaderRunner implements CommandLineRunner {
     @Override
     public void run(String... args) throws Exception {
         DataSourceUtil.setDB("oa");
-        Company company = companyService.getCompany();
-        Group defaultGroup = groupService.getDefaultGroup();
-        Map<String, Object> map = new HashMap<>();
-        map.put("company",company);
-        map.put("defaultGroup",defaultGroup);
-        // 使用DataService来存储数据
-        DataService.setConstantData(map);
+        try {
+            Company company = companyService.getCompany();
+            Group defaultGroup = groupService.getDefaultGroup();
+            Map<String, Object> map = new HashMap<>();
+            map.put("company", company);
+            map.put("defaultGroup", defaultGroup);
+            // 使用DataService来存储数据
+            DataService.setConstantData(map);
+        }catch (Exception ex){
+            System.err.println("获取OA默认数据失败");
+        }
         DataSourceUtil.setDB("v2");
     }
 }