修改内存去重改为存储最大时间

zork-server/influx-disorder/src/main/java/com/zork/common/service/InfluxDBService.java

@@ -25,7 +25,40 @@ public class InfluxDBService {
     }
 
     public QueryResult query(String command) {
-        return influxDB.query(new Query(command, database));
+        if (command == null || command.trim().isEmpty()) {
+            throw new IllegalArgumentException("查询命令不能为空");
+        }
+        if (influxDB == null) {
+            throw new IllegalStateException("InfluxDB连接未初始化");
+        }
+        
+        log.info("执行InfluxQL查询: {}", command);
+        long startTime = System.currentTimeMillis();
+        try {
+            QueryResult result = influxDB.query(new Query(command, database));
+            long duration = System.currentTimeMillis() - startTime;
+            log.info("查询执行完成，耗时: {}ms, 结果行数: {}", duration, getResultCount(result));
+            return result;
+        } catch (Exception e) {
+            long duration = System.currentTimeMillis() - startTime;
+            log.error("查询执行失败，耗时: {}ms, SQL: {}, 错误: {}", duration, command, e.getMessage());
+            throw e;
+        }
+    }
+
+    private int getResultCount(QueryResult result) {
+        try {
+            if (result != null && result.getResults() != null && !result.getResults().isEmpty()) {
+                QueryResult.Result firstResult = result.getResults().get(0);
+                if (firstResult.getSeries() != null && !firstResult.getSeries().isEmpty()) {
+                    QueryResult.Series series = firstResult.getSeries().get(0);
+                    return series.getValues() != null ? series.getValues().size() : 0;
+                }
+            }
+            return 0;
+        } catch (Exception e) {
+            return -1;
+        }
     }
 
     public void close() {

zork-server/influx-disorder/src/main/java/com/zork/disorder/component/BatchKafkaSender.java

@@ -11,7 +11,6 @@ import org.springframework.stereotype.Component;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -32,8 +31,6 @@ public class BatchKafkaSender {
     @Value("${spring.kafka.producer.topic}")
     private String topic;
     
-    @Value("${count-file}")
-    private String countFile;
     
     @Value("${batch.size:1000}")
     private int batchSize;
@@ -149,7 +146,7 @@ public class BatchKafkaSender {
         
         try {
             // 使用复用的Producer异步发送
-            producer.sendMessage(batch, countFile).whenComplete((result, throwable) -> {
+            producer.sendMessage(batch).whenComplete((result, throwable) -> {
                 if (throwable == null) {
                     // 线程安全更新统计信息
                     long newTotal = totalSent.addAndGet(batch.size());

zork-server/influx-disorder/src/main/java/com/zork/disorder/component/CountComponent.java

@@ -3,7 +3,6 @@ package com.zork.disorder.component;
 
 import com.zork.common.service.InfluxDBService;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
 import org.influxdb.dto.QueryResult;
 import org.springframework.beans.factory.annotation.Value;
@@ -21,7 +20,6 @@ import java.util.function.Predicate;
 import static com.zork.common.constant.InfluxDBConstant.AUTOGEN;
 import static com.zork.common.service.InfluxDBService.getResult;
 import static com.zork.common.utils.DateUtil.yearMonthDayBar;
-import static com.zork.common.utils.FileUtil.appendMethodB;
 
 /**
  * @Author: Prock.Liy
@@ -56,8 +54,6 @@ public class CountComponent {
     @Value("${interval-minute}")
     private Integer intervalMinute;
 
-    @Value("${count-file}")
-    private String countFile;
 
     @Value("${missing-s}")
     private Integer missingS;
@@ -120,9 +116,8 @@ public class CountComponent {
                             }
 
                             tableCount.put(simpleDateFormat.format(new Date()), series.getValues().get(0).get(1).toString());
-                            String ip = StringUtils.substringAfterLast(StringUtils.substringBeforeLast(url,":"),"/").replaceAll("\\.","_");
-                            // 用于统计每张表的总count
-                            appendMethodB(countFile + "\\" + ip + "_" + tableName + ".txt", tableCount.toString(), true);
+                            // 统计每张表的count（仅记录日志）
+                            log.info("表 {} 统计结果: {}", tableName, tableCount.toString());
                         });
                         // 更新count文件
                         influxDBService.close();
@@ -144,7 +139,6 @@ public class CountComponent {
      * @return
      */
     public Date lastFewMinutesMissingS(int minutes) {
-        long time = missingS * 1000;
         Calendar beforeTime = Calendar.getInstance();
         beforeTime.add(Calendar.MINUTE, -minutes);
         beforeTime.add(Calendar.SECOND, -missingS);

zork-server/influx-disorder/src/main/java/com/zork/disorder/component/InfluxDisorderComponent.java

@@ -12,15 +12,10 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
-import java.text.SimpleDateFormat;
 import java.util.*;
-import java.util.Objects;
-import java.util.stream.Collectors;
 
-import static com.zork.common.constant.InfluxDBConstant.AUTOGEN;
 import static com.zork.common.service.InfluxDBService.getResult;
 import static com.zork.common.utils.DateUtil.yearMonthDayBar;
-import static com.zork.common.utils.FileUtil.appendMethodB;
 
 /**
  * @Author: Prock.Liy
@@ -34,8 +29,6 @@ public class InfluxDisorderComponent {
     @Value("${table-keywords}")
     private String tableKeywords;
 
-    @Value("${count-file}")
-    private String countFile;
 
     @Value("${missing-s:15}")
     private Integer missingS;
@@ -45,31 +38,16 @@ public class InfluxDisorderComponent {
     private MultiDbConnectionManager connectionManager;
     
     @Autowired
-    private MultiDbTimestampTracker timestampTracker;
-    
-    @Autowired
-    private SmartDeduplicator deduplicator;
+    private DataTimestampIncrementalTracker dataTimestampTracker;
     
     @Autowired
     private BatchKafkaSender batchSender;
 
 
-    private final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
-    /**
-     * 每日凌晨清除当天统计的count条数
-     */
-    @Scheduled(cron = "${clearCountCron}")
-    public void clearCountData() {
-        try {
-            appendMethodB(countFile, "0", false);
-        } catch (Exception e) {
-            log.error("task Exception:{}", e.getMessage());
-        }
-    }
 
     /**
-     * 优化后的查询influxdb指标数据发送至Kafka（多库并行+去重+批量发送）
+     * 优化后的查询influxdb指标数据发送至Kafka（基于数据时间戳的增量查询+批量发送）
      */
     @Scheduled(cron = "${schedules}")
     public void optimizedExecute() {
@@ -79,16 +57,16 @@ public class InfluxDisorderComponent {
             List<String> urlList = connectionManager.getAvailableUrls();
             List<String> databaseList = connectionManager.getAvailableDatabases();
             
-            // 并行处理每个URL和数据库的组合
-            urlList.parallelStream().forEach(url -> {
-                databaseList.parallelStream().forEach(database -> {
+            // 顺序处理每个URL和数据库的组合，确保时间戳追踪准确
+            for (String url : urlList) {
+                for (String database : databaseList) {
                     try {
                         processUrlDatabase(url.trim(), database.trim());
                     } catch (Exception e) {
                         log.error("处理URL:{}, Database:{} 失败: {}", url, database, e.getMessage());
                     }
-                });
-            });
+                }
+            }
             
             log.info("InfluxDB数据采集任务完成");
             
@@ -115,12 +93,14 @@ public class InfluxDisorderComponent {
             
             log.info("URL:{}, Database:{} 找到 {} 张表需要处理", url, database, tableNameList.size());
             
-            // 并行处理每张表
-            List<NormalFieldsDTO> allData = tableNameList.parallelStream()
-                .map(tableName -> processTable(influxDBService, url, database, tableName))
-                .filter(Objects::nonNull)
-                .flatMap(List::stream)
-                .collect(Collectors.toList());
+            // 顺序处理每张表，保证时间戳更新的原子性
+            List<NormalFieldsDTO> allData = new ArrayList<>();
+            for (String tableName : tableNameList) {
+                List<NormalFieldsDTO> tableData = processTable(influxDBService, url, database, tableName);
+                if (tableData != null && !tableData.isEmpty()) {
+                    allData.addAll(tableData);
+                }
+            }
             
             // 批量发送到Kafka
             if (!allData.isEmpty()) {
@@ -134,16 +114,12 @@ public class InfluxDisorderComponent {
     }
     
     /**
-     * 处理单张表的数据
+     * 处理单张表的数据（基于数据时间戳的增量查询）
      */
     private List<NormalFieldsDTO> processTable(InfluxDBService influxDBService, String url, String database, String tableName) {
         try {
-            // 使用增量查询SQL
-            String sql = timestampTracker.buildDelayTolerantQuery(url, database, tableName);
-            if (sql == null) {
-                log.debug("表 {} 的查询时间窗口无效，跳过", tableName);
-                return Collections.emptyList();
-            }
+            // 使用基于数据时间戳的增量查询SQL
+            String sql = dataTimestampTracker.buildDataTimestampQuery(url, database, tableName);
             
             log.debug("查询表 {} SQL: {}", tableName, sql);
             QueryResult.Series series = getResult(influxDBService.query(sql));
@@ -153,8 +129,8 @@ public class InfluxDisorderComponent {
                 return Collections.emptyList();
             }
             
-            // 处理查询结果
-            return processTableData(series, url, database, tableName);
+            // 处理查询结果并更新最大时间戳
+            return processTableDataWithTimestampUpdate(series, url, database, tableName);
             
         } catch (Exception e) {
             log.error("处理表 {} 异常: {}", tableName, e.getMessage());
@@ -163,13 +139,14 @@ public class InfluxDisorderComponent {
     }
     
     /**
-     * 处理表数据，应用去重和转换
+     * 处理表数据，转换并更新时间戳（无需去重，增量查询天然避免重复）
      */
-    private List<NormalFieldsDTO> processTableData(QueryResult.Series series, String url, String database, String tableName) {
+    private List<NormalFieldsDTO> processTableDataWithTimestampUpdate(QueryResult.Series series, String url, String database, String tableName) {
         // 构建列名映射
         List<String> columns = series.getColumns();
         Map<String, Integer> columnIndexMap = buildColumnIndexMap(columns);
         
+        List<Atoota> atootaList = new ArrayList<>();
         List<NormalFieldsDTO> resultList = new ArrayList<>();
         
         for (List<Object> value : series.getValues()) {
@@ -180,10 +157,7 @@ public class InfluxDisorderComponent {
                     continue;
                 }
                 
-                // 应用智能去重
-                if (deduplicator.isDuplicate(url, database, tableName, atoota.getId())) {
-                    continue; // 跳过重复数据
-                }
+                atootaList.add(atoota);
                 
                 // 转换为DTO
                 NormalFieldsDTO dto = convertToDTO(atoota);
@@ -194,7 +168,10 @@ public class InfluxDisorderComponent {
             }
         }
         
-        log.debug("表 {} 处理完成，有效数据: {}", tableName, resultList.size());
+        // 更新最大时间戳（重要：只有数据处理成功后才更新）
+        int processedCount = dataTimestampTracker.updateMaxTimestamp(url, database, tableName, atootaList);
+        
+        log.debug("表 {} 处理完成，有效数据: {} 条", tableName, processedCount);
         return resultList;
     }
 
@@ -237,11 +214,11 @@ public class InfluxDisorderComponent {
     }
 
     /**
-     * 手动刷新所有缓存（用于重新开始同步）
+     * 手动重置时间戳（用于重新开始同步）
      */
-    public void resetAllCaches() {
-        deduplicator.clearAll();
-        log.info("已重置所有缓存");
+    public void resetAllTimestamps() {
+        dataTimestampTracker.clearAllTimestamps();
+        log.info("已重置所有时间戳");
     }
     
     /**
@@ -250,8 +227,8 @@ public class InfluxDisorderComponent {
     public String getSystemStats() {
         return String.format("系统状态 - %s, %s, %s", 
             connectionManager.getConnectionStats(),
-            timestampTracker.getStats(),
-            deduplicator.getCacheStats());
+            dataTimestampTracker.getStats(),
+            dataTimestampTracker.getCleanupStats());
     }
 
     /**
@@ -294,9 +271,10 @@ public class InfluxDisorderComponent {
     /**
      * 转换Atoota为DTO
      */
+    @SuppressWarnings("unchecked")
     private NormalFieldsDTO convertToDTO(Atoota atoota) {
         return NormalFieldsDTO.builder()
-            .normalFields(JSONObject.parseObject(JSONObject.toJSONString(atoota), Map.class))
+            .normalFields((Map<String, Object>) JSONObject.parseObject(JSONObject.toJSONString(atoota), Map.class))
             .timestamp(System.currentTimeMillis())
             .logTypeName(topic)
             .build();

zork-server/influx-disorder/src/main/java/com/zork/disorder/component/MultiDbConnectionManager.java

@@ -91,7 +91,13 @@ public class MultiDbConnectionManager {
         // 检查连接健康状态
         if (!isConnectionHealthy(connectionKey, service)) {
             log.warn("连接不健康，尝试重新创建 - URL: {}, Database: {}", url, database);
-            service = recreateConnection(url, database);
+            synchronized (this) {
+                // 双重检查，避免并发重复创建
+                service = connectionPool.get(connectionKey);
+                if (service == null || !isConnectionHealthy(connectionKey, service)) {
+                    service = recreateConnection(url, database);
+                }
+            }
         }
         
         return service;

zork-server/influx-disorder/src/main/java/com/zork/disorder/kafka/Producer.java

@@ -50,7 +50,7 @@ public class Producer {
     /**
      * 异步批量发送消息至Kafka - 无阻塞
      */
-    public CompletableFuture<Void> sendMessage(List<NormalFieldsDTO> normalFieldsDTOList, String countFile) {
+    public CompletableFuture<Void> sendMessage(List<NormalFieldsDTO> normalFieldsDTOList) {
         if (normalFieldsDTOList == null || normalFieldsDTOList.isEmpty()) {
             return CompletableFuture.completedFuture(null);
         }

zork-server/influx-disorder/src/main/resources/application.yml

-server:
-  port: 8190
-  tomcat:
-    uri-encoding: utf-8
-    basedir: "log/tomcat"
-    accesslog:
-      enabled: true
-      directory: "logs"
-      encoding: utf-8
-      file-date-format: _yyyy-MM-dd
-      locale: zh_CN
-      max-days: 30
-      prefix: "tomcat_access_log"
-      suffix: ".log"
-      pattern: "[${spring.application.name}] ===> 请求时间-%t 客户端IP-%a 客户端Host-%h
-      客户端协议-%H 线程名称-%I SessionId-%S 请求URL-%U 请求方法-%m 请求状态码-%s 耗时(ms)-%D 连接状态-%X"
-  shutdown: graceful
-
 spring:
   profiles:
     include:
       - datasource
       - kafka
 
-  jackson:
-    date-format: yyyy-MM-dd HH:mm:ss
-    time-zone: GMT+8
-  aop:
-    proxy-target-class: true
-  application:
-    name: "@artifactId@"
-    package-time: "@package-time@"
-
-# 优化后调度频率：30秒执行一次，给数据写入更多时间
-schedules: 0/30 * * * * ?
-
+schedules: 0/5 * * * * ?
+timestamp-file: ./data-timestamps.properties
+timestamp:
+  cleanup:
+    days: 3
 table-keywords: atoota
-
-send:
-  log-url:
-
-interval-minute: 1
-
-interval-second: 30
-
-initialDelay: 1
-
-# 优化后的配置
 missing-s: 15
-
-count-file: E:\\version
-
-clearCountCron: 0 0 0 * * ?
-
-# 批量处理配置  
 batch:
-  size: 1000       # 批次大小 - 减少积压风险
-  timeout: 2000   # 批次超时时间(ms) - 更快响应
-
+  size: 1000
+  timeout: 2000
\ No newline at end of file
-# InfluxDB多库配置示例
-# 多个URL用逗号分隔，多个数据库用逗号分隔
-# influxdb:
-#   url: "http://influx1:8086,http://influx2:8086"
-#   database: "metrics1,metrics2"
-
-logging:
-  level:
-    org.springframework.jdbc.core.JdbcTemplate: DEBUG
\ No newline at end of file

zork-server/influx-disorder/target/classes/application.yml

-server:
-  port: 8190
-  tomcat:
-    uri-encoding: utf-8
-    basedir: "log/tomcat"
-    accesslog:
-      enabled: true
-      directory: "logs"
-      encoding: utf-8
-      file-date-format: _yyyy-MM-dd
-      locale: zh_CN
-      max-days: 30
-      prefix: "tomcat_access_log"
-      suffix: ".log"
-      pattern: "[${spring.application.name}] ===> 请求时间-%t 客户端IP-%a 客户端Host-%h
-      客户端协议-%H 线程名称-%I SessionId-%S 请求URL-%U 请求方法-%m 请求状态码-%s 耗时(ms)-%D 连接状态-%X"
-  shutdown: graceful
-
 spring:
   profiles:
     include:
       - datasource
       - kafka
 
-  jackson:
-    date-format: yyyy-MM-dd HH:mm:ss
-    time-zone: GMT+8
-  aop:
-    proxy-target-class: true
-  application:
-    name: "@artifactId@"
-    package-time: "@package-time@"
-
-# 优化后调度频率：30秒执行一次，给数据写入更多时间
-schedules: 0/30 * * * * ?
-
+schedules: 0/5 * * * * ?
+timestamp-file: ./data-timestamps.properties
+timestamp:
+  cleanup:
+    days: 3
 table-keywords: atoota
-
-send:
-  log-url:
-
-interval-minute: 1
-
-interval-second: 30
-
-initialDelay: 1
-
-# 优化后的配置
 missing-s: 15
-
-count-file: E:\\version
-
-clearCountCron: 0 0 0 * * ?
-
-# 批量处理配置  
 batch:
-  size: 1000       # 批次大小 - 减少积压风险
-  timeout: 2000   # 批次超时时间(ms) - 更快响应
-
+  size: 1000
+  timeout: 2000
\ No newline at end of file
-# InfluxDB多库配置示例
-# 多个URL用逗号分隔，多个数据库用逗号分隔
-# influxdb:
-#   url: "http://influx1:8086,http://influx2:8086"
-#   database: "metrics1,metrics2"
-
-logging:
-  level:
-    org.springframework.jdbc.core.JdbcTemplate: DEBUG
\ No newline at end of file