refactor(service): 优化InfluxDB查询Flux构建逻辑

cn-universal-framework/cn-universal-dm/src/main/java/cn/universal/dm/device/service/log/InfluxDBDeviceLogService.java

@@ -85,7 +85,7 @@ public class InfluxDBDeviceLogService extends AbstractIoTDeviceLogService {
   @Value("${influxdb.url:http://127.0.0.1:8086}")
   private String url;
 
-  @Value("${influxdb.token}")
+  @Value("${influxdb.token:}")
   private String token;
 
   @Value("${influxdb.org:nexiot}")
@@ -670,22 +670,28 @@ public class InfluxDBDeviceLogService extends AbstractIoTDeviceLogService {
 
       // 1. 先查询总数（pivot后添加计数列，再进行count）
       // 关键：pivot后添加record_count=1列，然后对该列count
-      String countFlux = baseFilterFlux
-          + "  |> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")\n"
-          + "  |> group()\n"  // 合并所有分组
-          + "  |> map(fn: (r) => ({r with record_count: 1}))\n"  // 添加计数列
-          + "  |> count(column: \"record_count\")\n";  // 统计记录数
+      String countFlux =
+          baseFilterFlux
+              + "  |> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")\n"
+              + "  |> group()\n" // 合并所有分组
+              + "  |> map(fn: (r) => ({r with record_count: 1}))\n" // 添加计数列
+              + "  |> count(column: \"record_count\")\n"; // 统计记录数
       log.debug("InfluxDB查询元数据总数Flux: {}", countFlux);
       List<FluxTable> countTables = queryApi.query(countFlux);
       long total = parseCountResult(countTables);
 
       // 2. 再查询分页数据（包含pivot转换）
       int offset = (logQuery.getPageNum() - 1) * logQuery.getPageSize();
-      String dataFlux = baseFilterFlux
-          + "  |> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")\n"
-          + "  |> group()\n"
-          + "  |> sort(columns: [\"_time\"], desc: true)\n"
-          + "  |> limit(n: " + logQuery.getPageSize() + ", offset: " + offset + ")\n";
+      String dataFlux =
+          baseFilterFlux
+              + "  |> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")\n"
+              + "  |> group()\n"
+              + "  |> sort(columns: [\"_time\"], desc: true)\n"
+              + "  |> limit(n: "
+              + logQuery.getPageSize()
+              + ", offset: "
+              + offset
+              + ")\n";
       log.debug("InfluxDB查询元数据分页Flux: {}", dataFlux);
       List<FluxTable> dataTables = queryApi.query(dataFlux);
 
@@ -702,17 +708,13 @@ public class InfluxDBDeviceLogService extends AbstractIoTDeviceLogService {
     }
   }
 
-  /**
-   * 构建基础过滤条件（不含pivot、排序和分页）
-   */
+  /** 构建基础过滤条件（不含pivot、排序和分页） */
   private String buildBaseFilterFlux(LogQuery logQuery, String measurement) {
     StringBuilder flux = new StringBuilder();
     flux.append("from(bucket: \"").append(bucket).append("\")\n");
     // 核心改造：适配Unix时间戳的动态时间范围
     flux.append(buildRangeClause(logQuery)).append("\n");
-    flux.append("  |> filter(fn: (r) => r._measurement == \"")
-        .append(measurement)
-        .append("\")\n");
+    flux.append("  |> filter(fn: (r) => r._measurement == \"").append(measurement).append("\")\n");
 
     // 添加过滤条件
     if (StrUtil.isNotBlank(logQuery.getIotId())) {
@@ -739,9 +741,7 @@ public class InfluxDBDeviceLogService extends AbstractIoTDeviceLogService {
     return flux.toString();
   }
 
-  /**
-   * 解析count查询结果，获取总记录数
-   */
+  /** 解析count查询结果，获取总记录数 */
   private long parseCountResult(List<FluxTable> tables) {
     if (tables == null || tables.isEmpty()) {
       log.warn("Count查询返回空结果");

cn-universal-framework/cn-universal-dm/src/main/java/cn/universal/dm/device/service/log/MysqlDeviceLogService.java

@@ -17,12 +17,10 @@ import cn.hutool.core.util.CharsetUtil;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONObject;
-import cn.universal.common.constant.IoTConstant;
 import cn.universal.common.constant.IoTConstant.MessageType;
 import cn.universal.core.message.UPRequest;
 import cn.universal.core.metadata.AbstractPropertyMetadata;
 import cn.universal.dm.device.entity.IoTDevicePropertiesBO;
-import cn.universal.dm.device.service.impl.IoTDeviceService;
 import cn.universal.dm.device.service.impl.IoTProductDeviceService;
 import cn.universal.persistence.base.BaseUPRequest;
 import cn.universal.persistence.dto.IoTDeviceDTO;
@@ -37,17 +35,14 @@ import cn.universal.persistence.mapper.IoTDeviceLogMapper;
 import cn.universal.persistence.mapper.IoTDeviceLogMetadataMapper;
 import cn.universal.persistence.mapper.IoTDeviceLogMetadataShardMapper;
 import cn.universal.persistence.mapper.IoTDeviceLogShardMapper;
-import cn.universal.persistence.mapper.IoTDeviceMapper;
 import cn.universal.persistence.query.LogQuery;
 import cn.universal.persistence.query.PageBean;
 import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
 import jakarta.annotation.Resource;
 import java.util.List;
-import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
@@ -61,28 +56,19 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 public class MysqlDeviceLogService extends AbstractIoTDeviceLogService {
-
   private String storePolicy = "mysql";
-
   @Resource private IoTDeviceLogMapper ioTDeviceLogMapper;
   @Resource private IoTDeviceLogShardMapper ioTDeviceLogShardMapper;
-
   @Resource private IoTDeviceLogMetadataMapper ioTDeviceLogMetadataMapper;
   @Resource private IoTDeviceLogMetadataShardMapper ioTDeviceLogMetadataShardMapper;
-
-  @Resource private IoTDeviceMapper ioTDeviceMapper;
-
-  @Resource private IoTDeviceService iotDeviceService;
-
   @Resource private IoTProductDeviceService iotProductDeviceService;
-  @Resource private StringRedisTemplate stringRedisTemplate;
 
   /** 日志分表是否开启 */
-  @Value("${shard.log.enable}")
+  @Value("${shard.log.enable:true}")
   private Boolean enable;
 
   /** 日志meta分表是否开启 */
-  @Value("${shard.logMeta.enable}")
+  @Value("${shard.logMeta.enable:true}")
   private Boolean metaEnable;
 
   @Override
@@ -124,9 +110,6 @@ public class MysqlDeviceLogService extends AbstractIoTDeviceLogService {
     /** 产品数据存储策略，不为空则保存日志 */
     if (StrUtil.isNotBlank(ioTProduct.getStorePolicy())) {
       try {
-        //        ioTDeviceLog.setPoint(ioTDeviceDTO.getCoordinate());
-        //        ioTDeviceLogMapper.insertSelective(ioTDeviceLog);
-        // 日志分表 暂时双写单读
         if (enable) {
           ioTDeviceLogShardMapper.insertSelective(ioTDeviceLog);
         }
@@ -157,33 +140,9 @@ public class MysqlDeviceLogService extends AbstractIoTDeviceLogService {
                   IoTDeviceLogMetadataBuilder.ext1(ioTDevicePropertiesBO.getPropertyName());
                   IoTDeviceLogMetadataBuilder.ext2(ioTDevicePropertiesBO.getFormatValue());
                   IoTDeviceLogMetadataBuilder.ext3(ioTDevicePropertiesBO.getSymbol());
-                  // 新旧表都改
                   if (metaEnable) {
                     ioTDeviceLogMetadataShardMapper.insertUseGeneratedKeys(
                         IoTDeviceLogMetadataBuilder.build());
-                    Boolean re2 =
-                        stringRedisTemplate
-                            .opsForValue()
-                            .setIfAbsent(
-                                IoTConstant.LOG_META_SHARD_PROPERTY_DELETE_SIGN
-                                    + ":"
-                                    + up.getProductKey()
-                                    + ":"
-                                    + up.getDeviceId(),
-                                "1",
-                                20,
-                                TimeUnit.HOURS);
-                    if (Boolean.TRUE.equals(re2)) {
-                      Integer topId =
-                          ioTDeviceLogMetadataShardMapper.getTopPropertiesRecord(
-                              up.getIotId(),
-                              logStorePolicyDTO.getProperties().get(key).getMaxStorage(),
-                              key);
-                      if (topId != null) {
-                        ioTDeviceLogMetadataShardMapper.deleteTopPropertiesRecord(
-                            up.getIotId(), topId, key);
-                      }
-                    }
                   }
                 }
               });
@@ -199,27 +158,6 @@ public class MysqlDeviceLogService extends AbstractIoTDeviceLogService {
       IoTDeviceLogMetadataBuilder.content(up.getEventName());
       if (metaEnable) {
         ioTDeviceLogMetadataShardMapper.insertUseGeneratedKeys(IoTDeviceLogMetadataBuilder.build());
-        Boolean re2 =
-            stringRedisTemplate
-                .opsForValue()
-                .setIfAbsent(
-                    IoTConstant.LOG_META_SHARD_EVENT_DELETE_SIGN
-                        + ":"
-                        + up.getProductKey()
-                        + ":"
-                        + up.getDeviceId(),
-                    "1",
-                    20,
-                    TimeUnit.HOURS);
-        if (Boolean.TRUE.equals(re2)) {
-          Integer topId =
-              ioTDeviceLogMetadataShardMapper.getTopEventRecord(
-                  up.getIotId(), maxStorage, up.getEvent());
-          if (topId != null) {
-            ioTDeviceLogMetadataShardMapper.deleteTopEventRecord(
-                up.getIotId(), topId, up.getEvent());
-          }
-        }
       }
     }
   }
@@ -248,7 +186,6 @@ public class MysqlDeviceLogService extends AbstractIoTDeviceLogService {
   @Override
   public IoTDeviceLogVO queryById(LogQuery logQuery) {
     IoTDeviceLogVO ioTDeviceLogVO = ioTDeviceLogMapper.queryLogById(logQuery);
-    //    IoTDeviceLogVO ioTDeviceLogVO = BeanUtil.toBean(devLog, IoTDeviceLogVO.class);
     return ioTDeviceLogVO;
   }