修复 小麦赤霉病设备数据重复的问题

src/main/java/com/yunfeiyun/agmp/iots/service/IotYbqEnvDataService.java

@@ -2,6 +2,8 @@ package com.yunfeiyun.agmp.iots.service;
 
 import com.yunfeiyun.agmp.iot.common.domain.IotYbqEnvData;
 
+import java.util.List;
+
 /**
  * @author zhangn
  */
@@ -17,5 +19,5 @@ public interface IotYbqEnvDataService {
      */
     void deleteData(String id);
 
-
+    void insertBatchData(List<IotYbqEnvData> dataList);
 }

src/main/java/com/yunfeiyun/agmp/iots/service/IotYbqPredictDataService.java

@@ -2,6 +2,8 @@ package com.yunfeiyun.agmp.iots.service;
 
 import com.yunfeiyun.agmp.iot.common.domain.IotYbqPredictData;
 
+import java.util.List;
+
 /**
  * @author zhangn
  */
@@ -35,5 +37,19 @@ public interface IotYbqPredictDataService {
      */
     IotYbqPredictData getTodayData(String devBid);
 
+    /**
+     * 获取今天最新数据
+     *
+     * @param devBidList@return
+     */
+    List<IotYbqPredictData> getTodayLatestDataList(List<String> devBidList);
+
+    /**
+     * 批量插入预测数据
+     *
+     * @param dataList
+     */
+    void insertBatchData(List<IotYbqPredictData> dataList);
+
 }
 

src/main/java/com/yunfeiyun/agmp/iots/service/impl/IotYbqEnvDataServiceImpl.java

@@ -6,6 +6,8 @@ import com.yunfeiyun.agmp.iots.service.IotYbqEnvDataService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.List;
+
 /**
  * @author zhangn
  */
@@ -24,4 +26,9 @@ public class IotYbqEnvDataServiceImpl implements IotYbqEnvDataService {
     public void deleteData(String id) {
         mongoService.removeAllByParam("id",id,"IotYbqEnvData");
     }
+
+    @Override
+    public void insertBatchData(List<IotYbqEnvData> dataList) {
+        mongoService.insertList(IotYbqEnvData.class,dataList);
+    }
 }

src/main/java/com/yunfeiyun/agmp/iots/service/impl/IotYbqPredictDataServiceImpl.java

@@ -7,6 +7,7 @@ import com.yunfeiyun.agmp.iots.service.IotYbqPredictDataService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -44,9 +45,37 @@ public class IotYbqPredictDataServiceImpl implements IotYbqPredictDataService {
         param.put("devBid", devBid);
         param.put("date_ybqdataCreatedDate", start + " 00:00" + "," + start + " 23:59:59");
         List<IotYbqPredictData> iotYbqPredictDataList = mongoService.findAll(IotYbqPredictData.class, param);
-        if (iotYbqPredictDataList == null || iotYbqPredictDataList.size() == 0) {
+        if (iotYbqPredictDataList == null || iotYbqPredictDataList.isEmpty()) {
             return null;
         }
         return iotYbqPredictDataList.get(0);
     }
+
+    /**
+     * 获取今天最新数据
+     *
+     * @param devBidList@return
+     */
+    @Override
+    public List<IotYbqPredictData> getTodayLatestDataList(List<String> devBidList) {
+        String start = DateUtils.getDate();
+        Map<String, Object> param = new HashMap<>();
+        param.put("newList_devBid", devBidList);
+        param.put("timeStr_ybqdataCreatedDate", start + " 00:00" + "," + start + " 23:59:59");
+        List<IotYbqPredictData> iotYbqPredictDataList = mongoService.findAll(IotYbqPredictData.class, param, "ybqdataCreatedDate desc");
+        if (iotYbqPredictDataList == null || iotYbqPredictDataList.isEmpty()) {
+            iotYbqPredictDataList = new ArrayList<>();
+        }
+        return iotYbqPredictDataList;
+    }
+
+    /**
+     * 批量插入预测数据
+     *
+     * @param dataList
+     */
+    @Override
+    public void insertBatchData(List<IotYbqPredictData> dataList) {
+        mongoService.insertList(IotYbqPredictData.class, dataList);
+    }
 }

src/main/java/com/yunfeiyun/agmp/iots/task/YbqScheduler.java

@@ -9,21 +9,19 @@ import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
 import com.yunfeiyun.agmp.iot.common.domain.IotYbqEnvData;
 import com.yunfeiyun.agmp.iot.common.domain.IotYbqPredictData;
 import com.yunfeiyun.agmp.iot.common.enums.ybq.YbqTypeConst;
-import com.yunfeiyun.agmp.iots.core.manager.HttpManager;
-import com.yunfeiyun.agmp.iots.service.IotYbqEnvDataService;
-import com.yunfeiyun.agmp.iots.service.IotYbqPredictDataService;
 import com.yunfeiyun.agmp.iots.core.http.YbqCmbService;
+import com.yunfeiyun.agmp.iots.core.manager.HttpManager;
 import com.yunfeiyun.agmp.iots.service.IIotDeviceService;
 import com.yunfeiyun.agmp.iots.service.IIotDevicelasteddataService;
+import com.yunfeiyun.agmp.iots.service.IotYbqEnvDataService;
+import com.yunfeiyun.agmp.iots.service.IotYbqPredictDataService;
 import com.yunfeiyun.agmp.iots.warn.util.NumberUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
-import java.util.Arrays;
-import java.util.Date;
-import java.util.List;
+import java.util.*;
 
 /**
  * 作物xx虫子检测设备调度
@@ -63,6 +61,71 @@ public class YbqScheduler {
     };
 
 
+//    /**
+//     * 同步预测数据
+//     * 步骤：
+//     * 1. 遍历类型查询厂家设备
+//     * 2. 根据设备查询预测数据，入库
+//     * 每十分钟：0 0/10 0/1 * * ?
+//     * 每1小时：0 0 0/1 * * ?
+//     * 每天12点：0 0 8 * * ?
+//     * 每天8点：10分 0 10 8 * * ?
+//     * 对方：每天8点更新
+//     * 我们：每天8:10更新
+//     */
+//    @Scheduled(cron = "0 10 8 * * ?")
+//    public void synPredictedData() {
+//        String startDate = DateUtils.dateTime();
+//        log.info("【开始执行】同步预测数据任务，当前时间: {}", startDate);
+//        for (YbqTypeConst ybqTypeConst : ybqTypeConsts) {
+//            try {
+//                log.info("正在处理设备类型: {}", ybqTypeConst.getMessage());
+//                // 同步赤霉病设备预测数据
+//                List<IotDevice> iotDevices = getDevListByServiceName(ybqTypeConst.getServiceName());
+//                for (IotDevice iotDevice : iotDevices) {
+//                    delaYbqPredictedData(iotDevice, ybqTypeConst, startDate, startDate);
+//                }
+//            } catch (Exception e) {
+//                log.error("处理类型 {} 的设备预测数据时发生异常: ", ybqTypeConst.getMessage(), e);
+//            }
+//
+//        }
+//        log.info("【完成】同步预测数据任务执行完毕");
+//    }
+
+//    /**
+//     * 同步上报数据
+//     * 测试：每五分钟：0 0/5 0/1 * * ?
+//     * 对方：每2个小时 13、15
+//     * 2024-06-05 12:10:00
+//     * 2024-06-05 13:10:00
+//     * 2024-06-05 14:10:00
+//     * 2024-06-05 15:10:00
+//     * 2024-06-05 16:10:00
+//     * 延迟10分钟再拉取
+//     * 我们:从10分开始，每一个小时拉取一次
+//     */
+//    @Scheduled(cron = "0 10/59 0/1 * * ? ")
+//    //@Scheduled(cron = "0 0/2 0/1 * * ?")
+//    public void synReportData() {
+//        String startTime = DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, DateUtils.addMinutes(new Date(), -20));
+//        String endTime = DateUtils.dateTimeNow();
+//        log.info("【开始】同步上报数据任务执行于 {}", endTime);
+//        for (YbqTypeConst ybqTypeConst : ybqTypeConsts) {
+//            try {
+//                log.info("正在处理设备类型: {}", ybqTypeConst.getMessage());
+//                // 同步各类设备预测数据
+//                List<IotDevice> iotDevices = getDevListByServiceName(ybqTypeConst.getServiceName());
+//                for (IotDevice iotDevice : iotDevices) {
+//                    dealEnvDataByDevice(iotDevice, ybqTypeConst, startTime, endTime);
+//                }
+//            } catch (Exception e) {
+//                log.error("处理类型 {} 的设备预测数据时发生异常: ", ybqTypeConst.getMessage(), e);
+//
+//            }
+//        }
+//    }
+
     /**
      * 同步预测数据
      * 步骤：
@@ -76,6 +139,7 @@ public class YbqScheduler {
      * 我们：每天8:10更新
      */
     @Scheduled(cron = "0 10 8 * * ?")
+//    @Scheduled(cron = "*/10 * * * * ?")
     public void synPredictedData() {
         String startDate = DateUtils.dateTime();
         log.info("【开始执行】同步预测数据任务，当前时间: {}", startDate);
@@ -83,10 +147,7 @@ public class YbqScheduler {
             try {
                 log.info("正在处理设备类型: {}", ybqTypeConst.getMessage());
                 // 同步赤霉病设备预测数据
-                List<IotDevice> iotDevices = getDevListByServiceName(ybqTypeConst.getServiceName());
-                for (IotDevice iotDevice : iotDevices) {
-                    delaYbqPredictedData(iotDevice, ybqTypeConst, startDate, startDate);
-                }
+                updateDelaYbqPredictedData(ybqTypeConst, startDate, startDate);
             } catch (Exception e) {
                 log.error("处理类型 {} 的设备预测数据时发生异常: ", ybqTypeConst.getMessage(), e);
             }
@@ -96,78 +157,6 @@ public class YbqScheduler {
     }
 
     /**
-     * 根据设备处理预测数据
-     *
-     * @param iotDevice
-     * @param ybqTypeConst
-     * @param startDate
-     * @param endTime
-     */
-    public void delaYbqPredictedData(IotDevice iotDevice, YbqTypeConst ybqTypeConst, String startDate, String endTime) {
-        try {
-            String code = iotDevice.getDevCode();
-            String devBid = iotDevice.getDevBid();
-            JSONArray jsonArrayDevs = ((YbqCmbService) httpManager.getHttpClientByDevice(iotDevice)).getPredictedData(code, startDate, endTime, ybqTypeConst.getCode());
-            if (jsonArrayDevs == null || jsonArrayDevs.size() == 0) {
-                log.error("【预测数据同步】【{}}】暂无设备数据，无需导入", ybqTypeConst.getMessage());
-                return;
-            }
-            int count = jsonArrayDevs.size();
-            for (int i = 0; i < count; i++) {
-                try {
-                    JSONObject jobj = jsonArrayDevs.getJSONObject(i);
-                    if (jobj != null) {
-                        // 得到第三方返回的设备的预测数据
-                        String computeDate = jobj.getString("computeDate");
-                        String deviceId = jobj.getString("deviceId");
-                        String value = NumberUtil.keepTwoDecimalPlaces(jobj.getString("value"));
-                        //保存预测记录、更新预测最新数据
-                        // 查查当天的有没有，有的话更新，没有的话添加
-                        IotYbqPredictData iotYbqPredictDataToday = iotYbqPredictDataService.getTodayData(devBid);
-                        if (iotYbqPredictDataToday == null) {
-                            IotYbqPredictData iotYbqPredictData = new IotYbqPredictData();
-                            iotYbqPredictData.setId(IdUtils.fastUUID());
-                            iotYbqPredictData.setYbqdataBid(iotYbqPredictData.getId());
-                            iotYbqPredictData.setTid(iotDevice.getTid());
-                            iotYbqPredictData.setComputeDate(computeDate);
-                            iotYbqPredictData.setDateDevType(ybqTypeConst.getCode());
-                            iotYbqPredictData.setValue(value);
-                            iotYbqPredictData.setYbqdataContent(jobj);
-                            iotYbqPredictData.setDeviceId(deviceId);
-                            iotYbqPredictData.setDevBid(devBid);
-                            iotYbqPredictData.setDevTypeBid(iotDevice.getDevtypeBid());
-                            iotYbqPredictData.setYbqdataModifiedDate(DateUtils.dateTimeNow());
-                            iotYbqPredictData.setYbqdataCreatedDate(DateUtils.dateTimeNow());
-                            iotYbqPredictDataService.insertData(iotYbqPredictData);
-                        } else {
-                            iotYbqPredictDataToday.setComputeDate(computeDate);
-                            iotYbqPredictDataToday.setValue(value);
-                            iotYbqPredictDataToday.setYbqdataModifiedDate(DateUtils.dateTimeNow());
-                            // 只更新特定的字段
-                            iotYbqPredictDataService.updateData(iotYbqPredictDataToday);
-                        }
-                        // 同步更新设备信息
-                        String extInfo = iotDevice.getExtInfo() == null ? "{}" : iotDevice.getExtInfo();
-                        JSONObject jsonObject = JSONObject.parseObject(extInfo);
-                        //预测时间
-                        jsonObject.put("computeDate", DateUtils.dateTimeNow());
-                        //预测value
-                        jsonObject.put("computeValue", value);
-                        iIotDeviceService.updateIotDeviceExtInfo(devBid, jsonObject.toString());
-                        log.info("设备ID={} 的预测数据处理成功，computeDate={}, value={}", iotDevice.getDevBid(), computeDate, value);
-                    }
-                } catch (Exception e) {
-                    log.error("处理设备ID={} 的预测数据时出错: ", iotDevice.getDevBid(), e);
-                }
-            }
-        } catch (Exception e) {
-            log.error("处理类型 {} 的设备预测数据时发生异常: ", ybqTypeConst.getMessage(), e);
-        }
-
-    }
-
-
-    /**
      * 同步上报数据
      * 测试：每五分钟：0 0/5 0/1 * * ?
      * 对方：每2个小时 13、15
@@ -180,7 +169,6 @@ public class YbqScheduler {
      * 我们:从10分开始，每一个小时拉取一次
      */
     @Scheduled(cron = "0 10/59 0/1 * * ? ")
-    //@Scheduled(cron = "0 0/2 0/1 * * ?")
     public void synReportData() {
         String startTime = DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, DateUtils.addMinutes(new Date(), -20));
         String endTime = DateUtils.dateTimeNow();
@@ -188,16 +176,175 @@ public class YbqScheduler {
         for (YbqTypeConst ybqTypeConst : ybqTypeConsts) {
             try {
                 log.info("正在处理设备类型: {}", ybqTypeConst.getMessage());
-                // 同步各类设备预测数据
-                List<IotDevice> iotDevices = getDevListByServiceName(ybqTypeConst.getServiceName());
-                for (IotDevice iotDevice : iotDevices) {
-                    dealEnvDataByDevice(iotDevice, ybqTypeConst, startTime, endTime);
-                }
+                // 同步各类设备数据
+                updateEnvData(ybqTypeConst, startTime, endTime);
             } catch (Exception e) {
-                log.error("处理类型 {} 的设备预测数据时发生异常: ", ybqTypeConst.getMessage(), e);
+                log.error("处理类型 {} 的设备上报数据时发生异常: ", ybqTypeConst.getMessage(), e);
+            }
+        }
+    }
 
+    public List<IotYbqPredictData> getDelaYbqPredictedData(IotDevice iotDevice, YbqTypeConst ybqTypeConst, String startDate, String endDate){
+        String code = iotDevice.getDevCode();
+        String devBid = iotDevice.getDevBid();
+        List<IotYbqPredictData> iotYbqPredictDataList = new ArrayList<>();
+        try{
+            JSONArray jsonArrayDevs = ((YbqCmbService) httpManager.getHttpClientByDevice(iotDevice)).getPredictedData(code, startDate, endDate, ybqTypeConst.getCode());
+            if (jsonArrayDevs == null || jsonArrayDevs.isEmpty()) {
+                log.error("【预测数据同步】【{}}】暂无设备数据，无需导入", ybqTypeConst.getMessage());
+                return iotYbqPredictDataList;
+            }
+            for(int i = 0; i < jsonArrayDevs.size(); i++){
+                JSONObject jobj = jsonArrayDevs.getJSONObject(i);
+                // 得到第三方返回的设备的预测数据
+                String computeDate = jobj.getString("computeDate");
+                String deviceId = jobj.getString("deviceId");
+                String value = NumberUtil.keepTwoDecimalPlaces(jobj.getString("value"));
+                String createdDate = DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, DateUtils.dateTime("yyyyMMdd", computeDate));
+                IotYbqPredictData iotYbqPredictData = new IotYbqPredictData();
+                iotYbqPredictData.setId(IdUtils.fastUUID());
+                iotYbqPredictData.setYbqdataBid(iotYbqPredictData.getId());
+                iotYbqPredictData.setTid(iotDevice.getTid());
+                iotYbqPredictData.setComputeDate(computeDate);
+                iotYbqPredictData.setDateDevType(ybqTypeConst.getCode());
+                iotYbqPredictData.setValue(value);
+                iotYbqPredictData.setYbqdataContent(jobj);
+                iotYbqPredictData.setDeviceId(deviceId);
+                iotYbqPredictData.setDevBid(devBid);
+                iotYbqPredictData.setDevTypeBid(iotDevice.getDevtypeBid());
+                iotYbqPredictData.setYbqdataModifiedDate(createdDate);
+                iotYbqPredictData.setYbqdataCreatedDate(createdDate);
+
+                iotYbqPredictDataList.add(iotYbqPredictData);
             }
+        }catch (Exception e) {
+            log.error("获取设备ID={} 的预测数据时出错: ", iotDevice.getDevCode(), e);
         }
+        return iotYbqPredictDataList;
+    }
+
+    /**
+     * 根据设备处理上报数据
+     * @param iotDevice
+     * @param ybqTypeConst
+     * @param startDate
+     * @param endDate
+     * @return
+     */
+    public List<IotYbqEnvData> getDelaEnvData(IotDevice iotDevice, YbqTypeConst ybqTypeConst, String startDate, String endDate){
+        String code = iotDevice.getDevCode();
+        String devBid = iotDevice.getDevBid();
+        List<IotYbqEnvData> iotYbqEnvDataList = new ArrayList<>();
+        try{
+            JSONArray jsonArrayDevs = ((YbqCmbService) httpManager.getHttpClientByDevice(iotDevice)).getReportData(code, startDate, endDate);
+            if (jsonArrayDevs == null || jsonArrayDevs.isEmpty()) {
+                log.error("【上报数据同步】【{}}】暂无设备数据，无需导入", ybqTypeConst.getMessage());
+                return iotYbqEnvDataList;
+            }
+
+            for(int i = 0; i < jsonArrayDevs.size(); i++){
+                JSONObject jobj = jsonArrayDevs.getJSONObject(i);
+                if (jobj != null) {
+                    // 得到第三方返回的设备的预测数据
+                    IotYbqEnvData iotYbqEnvData = jobj.to(IotYbqEnvData.class);
+                    String collectTime = jobj.getString("collectTime");
+                    iotYbqEnvData.setId(IdUtils.fastUUID());
+                    iotYbqEnvData.setTid(iotDevice.getTid());
+                    iotYbqEnvData.setYbqdataBid(iotYbqEnvData.getId());
+                    iotYbqEnvData.setDevBid(devBid);
+                    iotYbqEnvData.setDateDevType(ybqTypeConst.getCode());
+                    iotYbqEnvData.setYbqdataContent(jobj);
+                    iotYbqEnvData.setYbqdataCreatedDate(collectTime);
+                    iotYbqEnvDataService.insertData(iotYbqEnvData);
+                    // 同步更新设备信息
+                    String extInfo = iotDevice.getExtInfo() == null ? "{}" : iotDevice.getExtInfo();
+                    JSONObject jsonObject = JSONObject.parseObject(extInfo);
+                    //上报时间
+                    jsonObject.put("collectTime", DateUtils.dateTimeNow());
+                    iIotDeviceService.updateIotDeviceExtInfo(devBid, jsonObject.toString());
+                    log.info("设备ID: {}, 收集时间: {} 的数据已成功插入数据库", devBid, collectTime);
+                    // 保存 设备最新数据 到redis
+                    iIotDevicelasteddataService.createOrUpdateDeviceLastedData(jobj, iotDevice, DateUtils.dateTimeNow(), 60 * 60 * 24L);
+                }
+            }
+        }catch (Exception e) {
+            log.error("获取设备ID={} 的上报数据时出错: ", iotDevice.getDevCode(), e);
+        }
+        return iotYbqEnvDataList;
+    }
+
+    /**
+     * 根据设备处理预测数据
+     *
+     * @param iotDevice
+     * @param ybqTypeConst
+     * @param startDate
+     * @param endTime
+     */
+    public void syncDelaYbqPredictedData(List<IotDevice> iotDevices, YbqTypeConst ybqTypeConst, String startDate, String endTime, boolean isCreate) {
+        try {
+            List<String> devBidList = new ArrayList<>();
+            for (IotDevice iotDevice : iotDevices) {
+                devBidList.add(iotDevice.getDevBid());
+            }
+            Map<String, IotYbqPredictData> iotYbqPredictDataMap = new HashMap<>();
+            // 查询今天最新的数据 根据预测时间
+            if(!isCreate){
+                List<IotYbqPredictData> iotYbqPredictDataList = iotYbqPredictDataService.getTodayLatestDataList(devBidList);
+                for(IotYbqPredictData iotYbqPredictData : iotYbqPredictDataList){
+                    String devBid = iotYbqPredictData.getDevBid();
+                    String computeDate = iotYbqPredictData.getComputeDate();
+                    String key = devBid + computeDate;
+                    iotYbqPredictDataMap.put(key,iotYbqPredictData);
+                }
+            }
+
+            List<IotYbqPredictData> insertList = new ArrayList<>();
+            for(IotDevice iotDevice : iotDevices){
+                List<IotYbqPredictData> dataList = getDelaYbqPredictedData(iotDevice, ybqTypeConst, startDate, endTime);
+                for(IotYbqPredictData iotYbqPredictData : dataList){
+                    String devBid = iotDevice.getDevBid();
+                    String computeDate = iotYbqPredictData.getComputeDate();
+                    String key = devBid + computeDate;
+                    if(!iotYbqPredictDataMap.containsKey(key)){
+                        insertList.add(iotYbqPredictData);
+                    }
+                }
+            }
+            log.info("需要插入的预测数据条数:{}", insertList.size());
+            log.info("需要插入的预测数据:{}",insertList);
+            if(!insertList.isEmpty()){
+                iotYbqPredictDataService.insertBatchData(insertList);
+            }
+        } catch (Exception e) {
+            log.error("处理类型 {} 的设备预测数据时发生异常: ", ybqTypeConst.getMessage(), e);
+        }
+    }
+
+    /**
+     * 创建预测数据
+     *
+     * @param iotDevice IoT设备对象
+     * @param ybqTypeConst YBQ类型常量
+     * @param startDate 开始时间
+     * @param endTime 结束时间
+     */
+    public void createDelaYbqPredictedData(IotDevice iotDevice, YbqTypeConst ybqTypeConst, String startDate, String endTime){
+        List<IotDevice> iotDevices = new ArrayList<>();
+        iotDevices.add(iotDevice);
+        syncDelaYbqPredictedData(iotDevices, ybqTypeConst, startDate, endTime, true);
+    }
+
+    /**
+     * 更新预测数据
+     *
+     * @param ybqTypeConst  ybq类型常量
+     * @param startDate     开始时间
+     * @param endTime       结束时间
+     */
+    public void updateDelaYbqPredictedData(YbqTypeConst ybqTypeConst, String startDate, String endTime){
+        List<IotDevice> iotDevices = getDevListByServiceName(ybqTypeConst.getServiceName());
+        syncDelaYbqPredictedData(iotDevices, ybqTypeConst, startDate, endTime, false);
     }
 
     /**
@@ -208,51 +355,81 @@ public class YbqScheduler {
      * @param startTime
      * @param endTime
      */
-    void dealEnvDataByDevice(IotDevice iotDevice, YbqTypeConst ybqTypeConst, String startTime, String endTime) {
+    void syncEnvDataByDevice(List<IotDevice> iotDevices, YbqTypeConst ybqTypeConst, String startTime, String endTime){
         try {
-            String code = iotDevice.getDevCode();
-            String devBid = iotDevice.getDevBid();
-            // 临时时间写成过去的，把数据拉起过来
-            JSONArray jsonArrayDevs = ((YbqCmbService) httpManager.getHttpClientByDevice(iotDevice)).getReportData(code, startTime, endTime);
-            if (jsonArrayDevs == null || jsonArrayDevs.size() == 0) {
-                log.error("【上报数据同步】【{}}】暂无设备数据，无需导入", ybqTypeConst.getMessage());
-                return;
-            }
-            int count = jsonArrayDevs.size();
-            for (int i = 0; i < count; i++) {
-                try {
-                    JSONObject jobj = jsonArrayDevs.getJSONObject(i);
-                    if (jobj != null) {
-                        // 得到第三方返回的设备的预测数据
-                        IotYbqEnvData iotYbqEnvData = jobj.to(IotYbqEnvData.class);
-                        String collectTime = jobj.getString("collectTime");
-                        iotYbqEnvData.setId(IdUtils.fastUUID());
+            List<IotYbqEnvData> insertList = new ArrayList<>();
+            Map<String, IotYbqEnvData> iotYbqEnvDataMap = new HashMap<>();
+            for(IotDevice iotDevice : iotDevices){
+                String devBid = iotDevice.getDevBid();
+                List<IotYbqEnvData> dataList = getDelaEnvData(iotDevice, ybqTypeConst, startTime, endTime);
+                String ybqdataCreatedDate = "0";
+                for(IotYbqEnvData iotYbqEnvData : dataList){
+                    String createdDate = iotYbqEnvData.getYbqdataCreatedDate();
+                    if(createdDate.compareTo(ybqdataCreatedDate) > 0){
+                        ybqdataCreatedDate = createdDate;
                         iotYbqEnvData.setTid(iotDevice.getTid());
-                        iotYbqEnvData.setYbqdataBid(iotYbqEnvData.getId());
-                        iotYbqEnvData.setDevBid(devBid);
-                        iotYbqEnvData.setDateDevType(ybqTypeConst.getCode());
-                        iotYbqEnvData.setYbqdataContent(jobj);
-                        //iotYbqEnvData.setYbqdataCreatedDate(DateUtils.dateTimeNow());
-                        iotYbqEnvData.setYbqdataCreatedDate(collectTime);
-                        iotYbqEnvDataService.insertData(iotYbqEnvData);
-                        // 同步更新设备信息
-                        String extInfo = iotDevice.getExtInfo() == null ? "{}" : iotDevice.getExtInfo();
-                        JSONObject jsonObject = JSONObject.parseObject(extInfo);
-                        //上报时间
-                        jsonObject.put("collectTime", DateUtils.dateTimeNow());
-                        iIotDeviceService.updateIotDeviceExtInfo(devBid, jsonObject.toString());
-                        log.info("设备ID: {}, 收集时间: {} 的数据已成功插入数据库", devBid, collectTime);
-                        // 保存 设备最新数据 到redis
-                        iIotDevicelasteddataService.createOrUpdateDeviceLastedData(jobj, iotDevice, DateUtils.dateTimeNow(), 60 * 60 * 24L);
+                        iotYbqEnvDataMap.put(devBid, iotYbqEnvData);
                     }
-                } catch (Exception e) {
-                    log.error("{}", e);
+                    insertList.add(iotYbqEnvData);
+                }
+                log.info("需要插入的上报数据条数:{}", insertList.size());
+                if(!insertList.isEmpty()){
+                    iotYbqEnvDataService.insertBatchData(insertList);
                 }
             }
-        } catch (Exception e) {
+            List<IotDevice> updateList = new ArrayList<>();
+            for(Map.Entry<String, IotYbqEnvData> entry : iotYbqEnvDataMap.entrySet()){
+                String devBid = entry.getKey();
+                IotYbqEnvData iotYbqEnvData = entry.getValue();
+                IotDevice updateIotDevice = new IotDevice();
+                updateIotDevice.setDevBid(devBid);
+                updateIotDevice.setTid(iotYbqEnvData.getTid());
+                updateIotDevice.setDevUpdateddate(iotYbqEnvData.getYbqdataCreatedDate());
+
+                updateList.add(updateIotDevice);
+            }
+
+            if(!updateList.isEmpty()){
+                iIotDeviceService.updateIotDeviceBatch(updateList);
+            }
+
+            for(IotDevice iotDevice : iotDevices){
+                String devBid = iotDevice.getDevBid();
+                IotYbqEnvData lastedData = iotYbqEnvDataMap.get(devBid);
+                if(lastedData != null){
+                    iIotDevicelasteddataService.createOrUpdateDeviceLastedData(lastedData.getYbqdataContent(), iotDevice, DateUtils.dateTimeNow(), 60 * 60 * 24L);
+                }
+            }
+
+        }catch (Exception e) {
             log.error("处理类型 {} 的设备上报数据时发生异常: ", ybqTypeConst.getMessage(), e);
         }
+    }
 
+    /**
+     * 创建设备数据
+     *
+     * @param iotDevice IoT设备对象
+     * @param ybqTypeConst YBQ类型常量
+     * @param startDate 开始时间
+     * @param endTime 结束时间
+     */
+    public void createEnvData(IotDevice iotDevice, YbqTypeConst ybqTypeConst, String startDate, String endTime){
+        List<IotDevice> iotDevices = new ArrayList<>();
+        iotDevices.add(iotDevice);
+        syncEnvDataByDevice(iotDevices, ybqTypeConst, startDate, endTime);
+    }
+
+    /**
+     * 更新设备数据
+     *
+     * @param ybqTypeConst  ybq类型常量
+     * @param startDate     开始时间
+     * @param endTime       结束时间
+     */
+    public void updateEnvData(YbqTypeConst ybqTypeConst, String startDate, String endTime){
+        List<IotDevice> iotDevices = getDevListByServiceName(ybqTypeConst.getServiceName());
+        syncEnvDataByDevice(iotDevices, ybqTypeConst, startDate, endTime);
     }
 
     /**
@@ -270,8 +447,6 @@ public class YbqScheduler {
 
         log.info("【开始】单独获取设备ID [{}] 的上报数据，用于首次激活", devId);
         try {
-
-
             String startTime = DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, DateUtils.addDays(new Date(), -30 * 3));
             String endTime = DateUtils.dateTimeNow();
 
@@ -279,31 +454,29 @@ public class YbqScheduler {
             String startDate = DateUtils.parseDateToStr("yyyyMMdd", DateUtils.addDays(new Date(), -30 * 3));
             String endDate = DateUtils.dateTime();
 
+            YbqTypeConst ybqTypeConst = null;
             // 处理赤霉病
             if (IotDeviceDictConst.TYPE_HS_YBQ_CMB.equals(iotDevice.getDevtypeBid())) {
                 log.info("处理设备ID [{}] 类型 [{}] 的监测环境数据和预测数据更新", devId, IotDeviceDictConst.TYPE_HS_YBQ_CMB);
-                // 更新监测环境数据
-                dealEnvDataByDevice(iotDevice, YbqTypeConst.YBQ_XM_CMB, startTime, endTime);
-                // 更新预测数据
-                delaYbqPredictedData(iotDevice, YbqTypeConst.YBQ_XM_CMB, startDate, endDate);
+                ybqTypeConst = YbqTypeConst.YBQ_XM_CMB;
             }
             // 处理稻瘟病
             else if (IotDeviceDictConst.TYPE_HS_YBQ_DWB.equals(iotDevice.getDevtypeBid())) {
                 log.info("处理设备ID [{}] 类型 [{}] 的监测环境数据和预测数据更新", devId, IotDeviceDictConst.TYPE_HS_YBQ_DWB);
-                // 更新检测环境数据
-                dealEnvDataByDevice(iotDevice, YbqTypeConst.YBQ_SD_DWB, startTime, endTime);
-                // 更新预测数据
-                delaYbqPredictedData(iotDevice, YbqTypeConst.YBQ_SD_DWB, startDate, endDate);
+                ybqTypeConst = YbqTypeConst.YBQ_SD_DWB;
             }  // 处理玉米大斑病
             else if (IotDeviceDictConst.TYPE_HS_YBQ_DBB.equals(iotDevice.getDevtypeBid())) {
                 log.info("处理设备ID [{}] 类型 [{}] 的监测环境数据和预测数据更新", devId, IotDeviceDictConst.TYPE_HS_YBQ_DBB);
-                // 更新检测环境数据
-                dealEnvDataByDevice(iotDevice, YbqTypeConst.YBQ_YM_DBB, startTime, endTime);
-                // 更新预测数据
-                delaYbqPredictedData(iotDevice, YbqTypeConst.YBQ_YM_DBB, startDate, endDate);
+                ybqTypeConst = YbqTypeConst.YBQ_YM_DBB;
             } else {
                 log.warn("设备ID [{}] 的设备类型未知，无法识别处理", devId);
             }
+            if(ybqTypeConst != null){
+                // 更新监测环境数据
+                createEnvData(iotDevice, ybqTypeConst, startDate, endDate);
+                // 插入预测数据
+                createDelaYbqPredictedData(iotDevice, ybqTypeConst, startDate, endDate);
+            }
         } catch (Exception e) {
             log.error("处理设备ID [{}] 上报数据时发生异常", devId, e);
         }