消息消费代码修改

data-storage/src/main/java/com/huaxing/data/mqtt/MqttMessageConsumer.java

@@ -35,7 +35,6 @@ public class MqttMessageConsumer extends MqttMessageReceiver {
     }
 
 
-
     @Override
     @Async("handleMessage")
     public void handleMessage(Message<?> message) throws MessagingException {
@@ -51,15 +50,9 @@ public class MqttMessageConsumer extends MqttMessageReceiver {
             if (topic.contains("iot/test1/in-storage")) {
                 // 模拟1000000条数据入库
                 log.info("接收到iot/test1/in-storage的消息啦，快去处理");
-                long l = System.currentTimeMillis();
+                CompletableFuture.runAsync(() -> {
-                System.currentTimeMillis();
-                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                     analysisService.parseStoreData(payload);
                 });
-//                future.get();
-                long l1 = System.currentTimeMillis();
-                log.info("入库完成，耗时：{}", l1 - l);
-                analysisService.parseStoreData(payload);
             } else if (topic.contains("table-update/")){} // TODO 表更新topic
         } catch (Exception e) {
             log.error(e.toString());

data-storage/src/main/java/com/huaxing/data/storage/service/impl/DataAnalysisService.java

@@ -26,7 +26,6 @@ public class DataAnalysisService implements IDataAnalysisService {
         this.dataStoredService = dataStoredService;
     }
 
-    int row = 0;
     /**
      * @author: swordmeng8@163.com
      * @date: 2025/1/13 11:16
@@ -40,7 +39,5 @@ public class DataAnalysisService implements IDataAnalysisService {
         log.info("入库数据解析完成");
         dataStoredService.insert(dataAnalysisDTO);
         log.info("入库数据入库完成");
-        row++;
-        log.info("入库数据入库完成，入库总数：{}", row);
     }
 }

data-storage/src/main/java/com/huaxing/test/TestController.java

@@ -119,6 +119,7 @@ public class TestController {
         map.put("deviceId", "0jZU2102_0806_0011");
         map.put("WM_WFA", 124.656 + index);
         map.put("WM_WFA_Unit", "m³");
+        map.put("test", "test");
         return map;
     }