优化日志查询

src/main/java/com/topsail/influxdb/controller/DeviceDataController.java

@@ -1,6 +1,7 @@
 package com.topsail.influxdb.controller;
 
 import com.topsail.influxdb.entity.DeviceHistoryData;
+import com.topsail.influxdb.entity.DeviceHistoryVo;
 import com.topsail.influxdb.result.Result;
 import com.topsail.influxdb.service.DeviceDataService;
 import org.slf4j.Logger;
@@ -12,6 +13,7 @@ import org.springframework.web.bind.annotation.*;
 import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 
 @RestController
@@ -29,11 +31,11 @@ public class DeviceDataController {
      * devices/history
      */
     @GetMapping(value = "/getShengDiLanHistory")
-    public Result<List<DeviceHistoryData>> getHistoryData(@RequestParam(name = "uid", required = false) String uid, @RequestParam(name = "pageNo", required = false) Integer pageNo, @RequestParam(name = "pageSize", required = false) Integer pageSize, @RequestParam(name = "startTime", required = false) String startTime, @RequestParam(name = "endTime", required = false) String endTime, @RequestParam(name = "imei") String imei) throws ParseException, BindException {
-        List<DeviceHistoryData> list = new ArrayList<>();
+    public Result getHistoryData(@RequestParam(name = "uid", required = false) String uid, @RequestParam(name = "pageNo", required = false) Integer pageNo, @RequestParam(name = "pageSize", required = false) Integer pageSize, @RequestParam(name = "startTime", required = false) String startTime, @RequestParam(name = "endTime", required = false) String endTime, @RequestParam(name = "imei") String imei) throws ParseException, BindException {
+        List<DeviceHistoryVo> list = new ArrayList<>();
         list = deviceDataService.getDeviceHistoryData(uid, pageNo, pageSize, startTime, endTime, imei);
-        logger.info("finished request " + uid);
-        return Result.success(list);
+        Map<String, Object> map = Result.returnPageMap(list);
+        return Result.success(map);
     }
 
     /**
@@ -42,7 +44,7 @@ public class DeviceDataController {
     @RequestMapping(value = "/shengDiLandelete", method = RequestMethod.GET)
     public Result<List<DeviceHistoryData>> deleteDeviceData() throws ParseException, BindException {
         List<DeviceHistoryData> list = new ArrayList<>();
-        deviceDataService.deleteDeviceData();
+//        deviceDataService.deleteDeviceData();
         return Result.success(list);
     }
 }

src/main/java/com/topsail/influxdb/controller/DeviceLogController.java

@@ -48,7 +48,7 @@ public class DeviceLogController {
      */
     @RequestMapping(value = "/shengDiLandelete", method = RequestMethod.GET)
     public Result delete() throws ParseException, BindException {
-        deviceLogService.deleteDeviceLog();
+//        deviceLogService.deleteDeviceLog();
         return Result.success(new CodeMsg(0, "success"));
     }
 }

src/main/java/com/topsail/influxdb/controller/HistoryDataController.java

@@ -30,7 +30,7 @@ public class HistoryDataController {
      * 转存设备历史数据（根据设备号查询设备历史数据）
      */
     @GetMapping(value = "/transferDeviceData")
-    public Result<List<DeviceHistoryData>> transferDeviceData(@RequestParam(name = "imei") String imei) throws ParseException, BindException {
+    public Result<List<DeviceHistoryData>> transferDeviceData(@RequestParam(name = "imei", required = false) String imei) throws ParseException, BindException {
          deviceDataService.transferDeviceData(imei);
         return Result.success(null);
     }
@@ -39,8 +39,8 @@ public class HistoryDataController {
      * 转存设备命令下发日志数据
      */
     @RequestMapping(value = "/transferDeviceLogData", method = RequestMethod.GET)
-    public Result<List<DeviceHistoryData>> transferDeviceLogData(@RequestParam(name = "imei", required = false) String imei) throws ParseException, BindException {
-        deviceLogService.transferDeviceLogData(imei);
+    public Result<List<DeviceHistoryData>> transferDeviceLogData(@RequestParam(name = "imei", required = false) String imei,@RequestParam (name = "companyId") Integer companyId) throws ParseException, BindException {
+        deviceLogService.transferDeviceLogData(imei,companyId);
         return Result.success(null);
     }
     /**

src/main/java/com/topsail/influxdb/entity/DeviceHistoryVo.java

@@ -0,0 +1,47 @@
+package com.topsail.influxdb.entity;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.format.annotation.DateTimeFormat;
+
+import java.util.Date;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+@JsonIgnoreProperties(value = {"hibernateLazyInitializer", "handler", "operations", "roles", "menus"})
+public class DeviceHistoryVo {
+
+    private Integer id; // id
+    private String deviceType; // 设备类型
+    private Integer supplierId; // 供应商id
+    private String imei; // 设备号
+    private Integer batteryLevel; // 电池电量
+    private Integer singalStrength; // 信号强度
+    private String sampleData; // 数据体
+    private String temperData; // 回水温度
+    private String humidityData; // 湿度
+    private String preessureData; // 压力
+    private Integer passNum; // 通道数
+    private Integer batteryState; // 电池状态
+    private String alarmType; // 告警类型
+    private String platformType; // 平台类型
+    private String userName; // 用户名
+    private String unit; // 单位
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
+    @DateTimeFormat(pattern = "yyyy-MM-dd")
+    private String sendTime; //  发送时间（设备发送时间）
+    private String dataBody; // 整个数据体
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
+    @DateTimeFormat(pattern = "yyyy-MM-dd")
+    private Date createtime; // 创建时间
+    private String value; // 数据值
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
+    @DateTimeFormat(pattern = "yyyy-MM-dd")
+    private Date sendDate; // 发送日期（实际接收时间）
+}

src/main/java/com/topsail/influxdb/mapper/DeviceInfoMapper.java

@@ -22,7 +22,8 @@ public interface DeviceInfoMapper {
      * @return
      */
 
-    List<SyncDataFlag> querySyncDeviceFlagInfo(@Param("imei") String imei);
+    List<SyncDataFlag> querySyncDeviceDataFlagInfo(@Param("imei") String imei);
+    List<SyncDataFlag> querySyncDeviceLogFlagInfo(@Param("imei") String imei);
 
     /**
      * 根据设备编号查询设备命令下发日志数据

src/main/java/com/topsail/influxdb/mapper/DeviceInfoMapper.xml

@@ -83,12 +83,14 @@
     </sql>
     <update id="updateSyncDeviceDataFlagInfo">
         UPDATE shengdilan_sync_data_flag
-        SET sync_device_data = #{flag}
+        SET sync_device_data = #{flag},
+            update_time = NOW()
         WHERE id = #{id}
     </update>
     <update id="updateSyncDeviceLogFlagInfo">
         UPDATE shengdilan_sync_data_flag
-        SET sync_device_log = #{flag}
+        SET sync_device_log = #{flag},
+            update_time = NOW()
         WHERE id = #{id}
     </update>
     <select id="queryDeviceBelongInfo" resultType="com.topsail.influxdb.entity.DeviceBelongInfo">
@@ -107,10 +109,19 @@
         WHERE FIND_IN_SET(#{imei}, h.valve_no)
         ORDER BY h.id DESC LIMIT 1
     </select>
-    <select id="querySyncDeviceFlagInfo" resultType="com.topsail.influxdb.entity.SyncDataFlag">
+    <select id="querySyncDeviceDataFlagInfo" resultType="com.topsail.influxdb.entity.SyncDataFlag">
         SELECT id, imei, sync_device_data, sync_device_log
         FROM shengdilan_sync_data_flag
-        WHERE 1=1
+        WHERE sync_device_data = 0
+        <if test="imei != null and imei!='' ">
+            AND imei = #{imei}
+        </if>
+
+    </select>
+    <select id="querySyncDeviceLogFlagInfo" resultType="com.topsail.influxdb.entity.SyncDataFlag">
+        SELECT id, imei, sync_device_data, sync_device_log
+        FROM shengdilan_sync_data_flag
+        WHERE sync_device_log = 0
         <if test="imei != null and imei!='' ">
             AND imei = #{imei}
         </if>

src/main/java/com/topsail/influxdb/rabbitmq/AmqpListener.java

@@ -49,7 +49,6 @@ public class AmqpListener {
             LOG.info("saveDeviceDataToInfluxdb OK:" + history.getImei());
             channel.basicAck(deliveryTag, false);
         } catch (Exception e) {
-            amqpService.SendMessage("shengdilandevicedataback", message);
             channel.basicNack(deliveryTag, false, true);
             LOG.info("saveDeviceDataToInfluxdb Error:(" + e + ")-" + message);
         }
@@ -76,7 +75,6 @@ public class AmqpListener {
             LOG.info("saveDeviceLogToInfluxdb OK:" + deviceLogData.getImei());
             channel.basicAck(deliveryTag, false);
         } catch (Exception e) {
-            amqpService.SendMessage("shengdilandevicelogback", message);
             channel.basicNack(deliveryTag, false, true);
             LOG.info("saveDeviceLogToInfluxdb Error:(" + e + ")-" + message);
         }

src/main/java/com/topsail/influxdb/service/DeviceDataService.java

@@ -9,10 +9,7 @@ import com.influxdb.client.WriteApi;
 import com.influxdb.client.domain.WritePrecision;
 import com.influxdb.query.FluxRecord;
 import com.influxdb.query.FluxTable;
-import com.topsail.influxdb.entity.DeviceBelongInfo;
-import com.topsail.influxdb.entity.DeviceDataInfluxData;
-import com.topsail.influxdb.entity.DeviceHistoryData;
-import com.topsail.influxdb.entity.SyncDataFlag;
+import com.topsail.influxdb.entity.*;
 import com.topsail.influxdb.mapper.DeviceInfoMapper;
 import com.topsail.influxdb.pojo.History;
 import com.topsail.influxdb.rabbitmq.service.AmqpService;
@@ -34,6 +31,7 @@ import java.util.*;
 @Service
 public class DeviceDataService {
     public static final Logger LOG = LoggerFactory.getLogger(DeviceDataService.class);
+    public static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
     // InfluxDB基础配置
     private static final String DEVICEDATA_BUCKET_NAME = "iot";
     @Value("${shengdilan.influxdb.token}")
@@ -66,7 +64,7 @@ public class DeviceDataService {
      * @param imei
      * @return
      */
-    public List<DeviceHistoryData> getDeviceHistoryData(String uid, Integer pageNo, Integer pageSize, String startTime, String endTime, String imei) {
+    public List<DeviceHistoryVo> getDeviceHistoryData(String uid, Integer pageNo, Integer pageSize, String startTime, String endTime, String imei) {
         InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org);
         StringBuffer query = new StringBuffer();
         query.append("from(bucket: \"iot\") ");
@@ -83,7 +81,7 @@ public class DeviceDataService {
                 e.printStackTrace();
             }
         } else {
-            query.append("|> range(start: -30d)");
+            query.append("|> range(start: -1d)");
         }
         query.append(String.format(" |> filter(fn: (r) => r[\"_measurement\"] == \"history\")  |> filter(fn: (r) => r[\"imei\"] == \"%s\")  |> filter(fn: (r) => r[\"_field\"] == \"jsondata\" ) |> sort(columns:[\"_time\"], desc:true)  ", imei));
         if (pageNo != null && pageSize != null) {
@@ -92,7 +90,8 @@ public class DeviceDataService {
         query.append(" |> yield(name: \"last\")");
         System.out.println("查询语句==========:" + query);
         List<FluxTable> tables = client.getQueryApi().query(query.toString());
-        List<DeviceHistoryData> rerurnList = new ArrayList<>();
+//        List<DeviceHistoryData> returnList = new ArrayList<>();
+        List<DeviceHistoryVo> resultSet = new ArrayList<>();
         for (FluxTable fluxTable : tables) {
             List<FluxRecord> records = fluxTable.getRecords();
             for (FluxRecord fluxRecord : records) {
@@ -100,13 +99,65 @@ public class DeviceDataService {
                 DeviceHistoryData data = JSONObject.parseObject(value, DeviceHistoryData.class);
                 Integer singalstrength = (Integer) JSONObject.parseObject(value).get("singalstrength");
                 data.setSignalStrength(singalstrength);
-                data.setTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format((Double.parseDouble(data.getTime()))));
-                rerurnList.add(data);
+                data.setTime(dateFormat.format((Long.parseLong(data.getTime()))));
+                DeviceHistoryVo dataVo = DeviceHistoryVo.builder()
+                        .deviceType(String.valueOf(data.getDeviceType()))
+                        .imei(data.getImei())
+                        .batteryLevel(data.getBatteryLevel())
+                        .singalStrength(data.getSignalStrength())
+                        .sampleData(data.getSampleData())
+                        .passNum(data.getPassNum())
+                        .alarmType(data.getAlarmType())
+                        .unit(data.getUnit())
+                        .sendTime(data.getTime())
+                        .dataBody(data.getDataBody())
+                        .value(data.getValue()).build();
+                dataVo = analysisSampleData(dataVo, data.getValue(), data.getUnit());
+                resultSet.add(dataVo);
+//                returnList.add(data);
             }
         }
         client.close();
-        Collections.reverse(rerurnList);
-        return rerurnList;
+        Collections.reverse(resultSet);
+        return resultSet;
+    }
+
+    /**
+     * 解析数据
+     *
+     * @param dataVo
+     * @param sampleData
+     * @param unit
+     * @return
+     */
+    private DeviceHistoryVo analysisSampleData(DeviceHistoryVo dataVo, String sampleData, String unit) {
+        if (sampleData != null && !"".equals(sampleData) && unit != null && !"".equals(unit)) {
+            //如果字符串中包含中括号则去掉中括号
+            if (sampleData.contains("[")) {
+                sampleData = sampleData.substring(sampleData.indexOf("[") + 1, sampleData.indexOf("]"));
+            }
+            String[] data = sampleData.split(",");
+            if (unit.contains("[")) {
+                unit = unit.substring(unit.indexOf("[") + 1, unit.indexOf("]"));
+            }
+            String[] unitString = unit.split(",");
+            for (int i = 0; i < unitString.length; i++) {
+                if (unitString[i].contains("℃") && data.length > i) {
+                    //温度
+                    dataVo.setTemperData(data[i]);
+                } else if (unitString[i].contains("O/F") && data.length > i) {
+                    //开度
+                    dataVo.setSampleData(data[i]);
+                } else if (unitString[i].contains("%RH") && data.length > i) {
+                    //湿度
+                    dataVo.setHumidityData(data[i]);
+                } else if (unitString[i].contains("MPa") && data.length > i) {
+                    //压力
+                    dataVo.setPreessureData(data[i]);
+                }
+            }
+        }
+        return dataVo;
     }
 
     /**
@@ -114,7 +165,7 @@ public class DeviceDataService {
      */
     public List<DeviceDataInfluxData> getOldInfluxdbData(String imei) {
         DeviceBelongInfo deviceBelongInfo = deviceInfoMapper.queryDeviceBelongInfo(imei);
-        InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org);
+        InfluxDBClient client = InfluxDBClientFactory.create(oldurl, oldtoken.toCharArray(), oldorg);
         StringBuffer query = new StringBuffer();
         query.append("from(bucket: \"iot\") ");
         query.append("|> range(start: -1y)");
@@ -183,8 +234,12 @@ public class DeviceDataService {
         DeviceDataInfluxData deviceDataInfluxData = rebuildDeviceDataInfluxData(history, time);
         try (WriteApi writeApi = client.getWriteApi()) {
             writeApi.writeMeasurement(DEVICEDATA_BUCKET_NAME, org, WritePrecision.NS, deviceDataInfluxData);
+        } catch (Exception e) {
+            e.printStackTrace();
+            LOG.error("设备数据写入influxdb失败：{}", history.getImei());
+            amqpService.SendMessage("shengdilandevicedataback", JSON.toJSONString(history));
         }
-        System.out.println("设备数据写入成功==========》" + history.getImei());
+        LOG.info("设备数据写入influxdb成功：{}", history.getImei());
     }
 
     /**
@@ -268,18 +323,18 @@ public class DeviceDataService {
      */
     public void transferDeviceData(String imei) {
         //1.查询所有的设备编号
-        List<SyncDataFlag> syncDataFlags = deviceInfoMapper.querySyncDeviceFlagInfo(imei);
+        List<SyncDataFlag> syncDataFlags = deviceInfoMapper.querySyncDeviceDataFlagInfo(imei);
         //2.根据设备编号查询历史Influxdb所有的设备数据
         if (syncDataFlags != null && syncDataFlags.size() > 0) {
             for (SyncDataFlag syncDataFlag : syncDataFlags) {
                 Boolean syncDeviceData = false;
-                List<DeviceDataInfluxData> influxdbDataList = getOldInfluxdbData(syncDataFlag.getImei());
+                List<DeviceDataInfluxData> influxdbDataList = getOldInfluxdbData(syncDataFlag.getImei().toLowerCase(Locale.ROOT));
                 if (influxdbDataList != null && influxdbDataList.size() > 0) {
                     for (DeviceDataInfluxData influxData : influxdbDataList) {
-                        InfluxDBClient client = InfluxDBClientFactory.create(oldurl, oldtoken.toCharArray());
+                        InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray());
                         //3.将设备数据保存到influxdb中
                         try (WriteApi writeApi = client.getWriteApi()) {
-                            writeApi.writeMeasurement(DEVICEDATA_BUCKET_NAME, oldorg, WritePrecision.NS, influxData);
+                            writeApi.writeMeasurement(DEVICEDATA_BUCKET_NAME, org, WritePrecision.NS, influxData);
                             syncDeviceData = true;
                         } catch (Exception e) {
                             LOG.error("保存设备数据到influxdb失败：{}", e.getMessage());

src/main/java/com/topsail/influxdb/service/DeviceLogService.java

@@ -50,6 +50,7 @@ public class DeviceLogService {
     DeviceInfoMapper deviceInfoMapper;
     @Autowired
     AmqpService amqpService;
+
     /**
      * 存储设备数据到influxdb
      *
@@ -80,8 +81,11 @@ public class DeviceLogService {
         deviceLogInfluxData.time = time != null ? time : Instant.now();
         try (WriteApi writeApi = client.getWriteApi()) {
             writeApi.writeMeasurement(LOG_BUCKET_NAME, org, WritePrecision.NS, deviceLogInfluxData);
+        } catch (Exception e) {
+            LOG.error("下发日志写入influxdb失败：{}", e.getMessage());
+            amqpService.SendMessage("shengdilandevicelogback", JSON.toJSONString(deviceLogData));
         }
-        System.out.println("下发日志写入成功==========》" + deviceLogData.getImei());
+        LOG.info("下发日志写入influxdb成功：{}", deviceLogData.getImei());
     }
 
     /**
@@ -110,6 +114,7 @@ public class DeviceLogService {
 
 
 //     from(bucket: "devicelog") |> range(start: -3600d) |> filter(fn: (r) => r["_measurement"] == "devicelog") |> filter(fn: (r) => r["_field"] == "logjson") |> filter(fn: (r) => r["deviceBelongInfo"] =~ /.*白志.*/) |> sort(columns:["_time"], desc:true) |> limit(n: 10, offset: 0) |> yield(name: "data")
+
     /**
      * 查询设备下发命令日志信息
      *
@@ -131,7 +136,7 @@ public class DeviceLogService {
         // 1. 拼接公共过滤条件片段（总条数和分页查询共用）
         StringBuilder filterFragment = new StringBuilder();
         filterFragment.append("from(bucket: \"devicelog\") ");
-        if (startTime != null && endTime != null) {
+        if (startTime != null && startTime != "" && endTime != null && endTime != "") {
             SimpleDateFormat oldFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
             SimpleDateFormat newFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
             newFormat.setTimeZone(TimeZone.getTimeZone("UTC"));//时区转换
@@ -144,7 +149,7 @@ public class DeviceLogService {
                 e.printStackTrace();
             }
         } else {
-            filterFragment.append("|> range(start: -3600d)");
+            filterFragment.append("|> range(start: -360d)");
         }
         // 2. 拼接固定过滤条件（_measurement和_field，必选）
         filterFragment.append(" |> filter(fn: (r) => r[\"_measurement\"] == \"devicelog\")");
@@ -198,10 +203,10 @@ public class DeviceLogService {
         dataQuery.append(filterFragment); // 复用过滤条件
         dataQuery.append(" |> sort(columns:[\"_time\"], desc:true)");
         // 处理分页参数默认值
-        int finalPageNum = (pageNode == null || pageNode <= 0) ? DEFAULT_PAGE_NUM : pageNode;
-        int finalPageSize = (pageSize == null || pageSize <= 0) ? DEFAULT_PAGE_SIZE : pageSize;
+        int finalPageNum = pageNode != null  ? pageNode:DEFAULT_PAGE_NUM;
+        int finalPageSize = pageSize != null ?  pageSize:DEFAULT_PAGE_SIZE;
         int offset = (finalPageNum - 1) * finalPageSize;
-        dataQuery.append(String.format(" |> limit(n: %d, offset: %d)", finalPageSize, offset));
+//        dataQuery.append(" |> limit(n: " + finalPageSize + ", offset: " + offset + ")");
         dataQuery.append(" |> yield(name: \"data\")"); // 标记结果集为data
         System.out.println("查询数量语句：" + totalQuery.toString());
         System.out.println("查询数据语句：" + dataQuery.toString());
@@ -230,6 +235,8 @@ public class DeviceLogService {
         client.close();
         Collections.reverse(rerurnList);
         resultSet.put("count", totalCount);
+        //对结果按照分页要求截取构造数据
+        rerurnList = rerurnList.subList(offset, Math.min(offset + finalPageSize, rerurnList.size()));
         resultSet.put("list", rerurnList);
         return resultSet;
     }
@@ -254,15 +261,15 @@ public class DeviceLogService {
     /**
      * 转存设备命令下发日志数据
      */
-    public void transferDeviceLogData(String searchImei) {
+    public void transferDeviceLogData(String searchImei, Integer companyId) {
         //1. 查询所有的设备编号
-        List<SyncDataFlag> syncDataFlags = deviceInfoMapper.querySyncDeviceFlagInfo(searchImei);
+        List<SyncDataFlag> syncDataFlags = deviceInfoMapper.querySyncDeviceLogFlagInfo(searchImei);
         if (syncDataFlags != null && syncDataFlags.size() > 0) {
             //2. 根据设备编号查询设备命令下发日志数据
             for (SyncDataFlag syncDataFlag : syncDataFlags) {
                 Boolean flag = false;
                 String imei = syncDataFlag.getImei();
-                List<DeviceLogData> deviceLogDataList = deviceInfoMapper.queryDeviceLogData(imei, 10);
+                List<DeviceLogData> deviceLogDataList = deviceInfoMapper.queryDeviceLogData(imei, companyId);
                 if (deviceLogDataList != null && deviceLogDataList.size() > 0) {
                     DeviceBelongInfo deviceBelongInfo = deviceInfoMapper.queryDeviceBelongInfo(imei);
                     //3. 批量插入设备命令下发日志数据到influxdb中
@@ -275,7 +282,7 @@ public class DeviceLogService {
                         } catch (Exception e) {
                             LOG.error("保存设备命令下发日志数据到influxdb中失败：" + e.getMessage());
                             flag = false;
-                            if(deviceLogData!=null ){
+                            if (deviceLogData != null) {
                                 amqpService.SendMessage("shengdilandevicelogback", JSON.toJSONString(deviceLogData));
                             }
                         }