fix: influxdb数据库

feat: 新增influxdb数据库
2026-03-24 14:47:52 +08:00 · 2026-03-24 13:50:50 +08:00
14 changed files with 806 additions and 10 deletions
--- a/nl-iot/pom.xml
+++ b/nl-iot/pom.xml
@@ -28,35 +28,36 @@
        <dependency>
            <groupId>org.eclipse.milo</groupId>
            <artifactId>sdk-client</artifactId>
-            <version>0.6.16</version>
        </dependency>

        <!-- Java COM opc da 相关 -->
        <dependency>
            <groupId>org.jinterop</groupId>
            <artifactId>j-interop</artifactId>
-            <version>2.0.4</version>
        </dependency>

        <dependency>
            <groupId>org.apache.plc4x</groupId>
            <artifactId>plc4j-api</artifactId>
-            <version>0.13.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.plc4x</groupId>
            <artifactId>plc4j-driver-s7</artifactId>
-            <version>0.13.0</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.plc4x</groupId>
            <artifactId>plc4j-driver-modbus</artifactId>
-            <version>0.13.0</version>
            <scope>runtime</scope>
        </dependency>
+
+        <!-- InfluxDB 2.x 官方Java客户端 -->
+        <dependency>
+            <groupId>com.influxdb</groupId>
+            <artifactId>influxdb-client-java</artifactId>
+        </dependency>
    </dependencies>

 </project>
--- a/nl-iot/src/main/java/org/nl/iot/config/AsyncConfig.java
+++ b/nl-iot/src/main/java/org/nl/iot/config/AsyncConfig.java
@@ -0,0 +1,56 @@
+package org.nl.iot.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.AsyncConfigurer;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * 异步任务配置
+ * 
+ * @author lyd
+ * @date 2026/03/24
+ */
+@Slf4j
+@Configuration
+public class AsyncConfig implements AsyncConfigurer {
+
+    /**
+     * InfluxDB异步写入线程池
+     */
+    @Bean("influxDbAsyncExecutor")
+    public Executor influxDbAsyncExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        // 核心线程数
+        executor.setCorePoolSize(2);
+        // 最大线程数
+        executor.setMaxPoolSize(8);
+        // 队列容量
+        executor.setQueueCapacity(200);
+        // 线程名前缀
+        executor.setThreadNamePrefix("InfluxDB-Async-");
+        // 线程空闲时间
+        executor.setKeepAliveSeconds(60);
+        // 拒绝策略：调用者运行策略
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        // 等待所有任务结束后再关闭线程池
+        executor.setWaitForTasksToCompleteOnShutdown(true);
+        // 等待时间
+        executor.setAwaitTerminationSeconds(60);
+        executor.initialize();
+        
+        log.info("InfluxDB异步线程池初始化完成 - 核心线程数: {}, 最大线程数: {}, 队列容量: {}", 
+            executor.getCorePoolSize(), executor.getMaxPoolSize(), executor.getQueueCapacity());
+        
+        return executor;
+    }
+
+    @Override
+    public Executor getAsyncExecutor() {
+        return influxDbAsyncExecutor();
+    }
+}
--- a/nl-iot/src/main/java/org/nl/iot/core/cache/MetadataCacheManager.java
+++ b/nl-iot/src/main/java/org/nl/iot/core/cache/MetadataCacheManager.java
@@ -1,11 +1,13 @@
 package org.nl.iot.core.cache;

+import cn.hutool.core.util.ObjectUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.nl.iot.core.driver.bo.ConnectSiteDO;
 import org.nl.iot.core.driver.bo.DeviceConnectionBO;
 import org.nl.iot.core.driver.bo.DeviceInfoReadCacheDo;
 import org.nl.iot.core.driver.bo.SiteBO;
 import org.nl.iot.core.driver.entity.RValue;
+import org.nl.iot.modular.influxdb.service.PlcSignalService;
 import org.nl.iot.modular.iot.service.IotConfigService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
@@ -29,6 +31,9 @@ public class MetadataCacheManager implements CommandLineRunner {
    @Autowired
    private IotConfigService configService;

+    @Autowired
+    private PlcSignalService plcSignalService;
+
    private final Map<String, DeviceInfoReadCacheDo> deviceInfoCache = new ConcurrentHashMap<>();
    
    /**
@@ -183,6 +188,14 @@ public class MetadataCacheManager implements CommandLineRunner {

    // ==================== 信号值缓存方法 ====================

+    /**
+     * 存储单个信号值
+     * @param key connectCode.deviceCode.信号别名
+     * @param rValue 信号值对象
+     */
+    public void putSignalValue(String key, RValue rValue) {
+        signalValueCache.put(key, rValue);
+    }
    /**
     * 存储单个信号值
     * @param deviceKey connectCode.deviceCode
@@ -203,12 +216,27 @@ public class MetadataCacheManager implements CommandLineRunner {
        if (rValues == null || rValues.isEmpty()) {
            return;
        }
-        
-        for (RValue rValue : rValues) {
-            if (rValue != null && rValue.getSiteBO() != null) {
-                String alias = rValue.getSiteBO().getAlias();
-                putSignalValue(deviceKey, alias, rValue);
+        Map<String, RValue> cacheSnapshot = getAllSignalValues();
+        List<RValue> needSaveValues = new ArrayList<>();
+        for (RValue newRValue : rValues) {
+            if (newRValue != null && newRValue.getSiteBO() != null) {
+                String alias = newRValue.getSiteBO().getAlias();
+                String key = deviceKey + "." + alias;
+                RValue oldValue = cacheSnapshot.get(key);
+                if (ObjectUtil.isNotEmpty(oldValue) && oldValue.getValue().equals(newRValue.getValue())) {
+                    continue;
                }
+                putSignalValue(key, newRValue);
+                needSaveValues.add(newRValue);
+            }
+        }
+        // 异步批量存储到influxDB
+        if (!needSaveValues.isEmpty()) {
+            plcSignalService.batchSaveSignalToInfluxDBAsync(needSaveValues)
+                .exceptionally(throwable -> {
+                    log.error("异步保存信号到InfluxDB失败，设备: {}, 数量: {}", deviceKey, needSaveValues.size(), throwable);
+                    return null;
+                });
        }
    }

--- a/nl-iot/src/main/java/org/nl/iot/core/influxDB/InfluxDB2Config.java
+++ b/nl-iot/src/main/java/org/nl/iot/core/influxDB/InfluxDB2Config.java
@@ -0,0 +1,83 @@
+package org.nl.iot.core.influxDB;
+
+import com.influxdb.client.*;
+import jakarta.annotation.PreDestroy;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * influx2.0
+ * @Author: liyongde
+ * @Date: 2026/3/24 10:38
+ * @Modified By:
+ */
+@Configuration
+public class InfluxDB2Config {
+
+    @Value("${influx.url}")
+    private String url;
+
+    @Value("${influx.token}")
+    private String token;
+
+    @Value("${influx.org}")
+    private String org;
+
+    @Value("${influx.bucket}")
+    private String bucket;
+
+    private InfluxDBClient influxDBClient;
+
+    /**
+     * 初始化InfluxDB 2.x客户端
+     * 修复：通过OkHttpClient配置超时，替代不存在的set超时方法
+     */
+    @Bean
+    public InfluxDBClient influxDBClient() {
+
+        // 创建客户端，传入自定义OkHttp配置
+        this.influxDBClient = InfluxDBClientFactory.create(
+                url,
+                token.toCharArray(),
+                org,
+                bucket
+        );
+
+        // 测试连接
+        influxDBClient.ping();
+        return influxDBClient;
+    }
+
+    /**
+     * 写入API（开启批量写入，提升性能）
+     */
+    @Bean
+    public WriteApi writeApi(InfluxDBClient client) {
+        // 批量写入配置：缓冲1000条、500ms刷新、最大5000条
+        WriteOptions writeOptions = WriteOptions.builder()
+                .batchSize(1000)
+                .flushInterval(500)
+                .bufferLimit(5000)
+                .build();
+        return client.makeWriteApi(writeOptions);
+    }
+
+    /**
+     * 查询API
+     */
+    @Bean
+    public QueryApi queryApi(InfluxDBClient client) {
+        return client.getQueryApi();
+    }
+
+    /**
+     * 项目关闭时销毁客户端
+     */
+    @PreDestroy
+    public void destroy() {
+        if (influxDBClient != null) {
+            influxDBClient.close();
+        }
+    }
+}
--- a/nl-iot/src/main/java/org/nl/iot/core/influxDB/InfluxDB2Util.java
+++ b/nl-iot/src/main/java/org/nl/iot/core/influxDB/InfluxDB2Util.java
@@ -0,0 +1,105 @@
+package org.nl.iot.core.influxDB;
+
+import com.influxdb.client.QueryApi;
+import com.influxdb.client.WriteApi;
+import com.influxdb.client.domain.WritePrecision;
+import com.influxdb.query.FluxRecord;
+import com.influxdb.query.FluxTable;
+import org.nl.iot.modular.influxdb.entity.PlcSignal;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 工具类
+ * @Author: liyongde
+ * @Date: 2026/3/24 10:43
+ * @Modified By:
+ */
+@Component
+public class InfluxDB2Util {
+
+    @Value("${influx.bucket}")
+    private String bucket;
+
+    @Value("${influx.org}")
+    private String org;
+
+    @Autowired
+    private WriteApi writeApi;
+
+    @Autowired
+    private QueryApi queryApi;
+
+    /**
+     * 批量写入PLC信号
+     */
+    public void insertBatch(List<PlcSignal> datas) {
+        writeApi.writeMeasurements(WritePrecision.NS, datas);
+    }
+
+
+    /**
+     * 单条写入PLC信号
+     */
+    public void insertOne(PlcSignal data) {
+        insertOne(bucket, org, data);
+    }
+
+    /**
+     * 单条写入PLC信号
+     */
+    public void insertOne(String bucket, String org, PlcSignal data) {
+        writeApi.writeMeasurement(bucket, org, WritePrecision.NS, data);
+    }
+
+    /**
+     * PLC专用Flux查询结果转换
+     */
+    public List<PlcSignal> queryPlcFlux(String flux) {
+        List<FluxTable> tables = queryApi.query(flux);
+        return tables.stream()
+                .flatMap(table -> table.getRecords().stream())
+                .map(this::convertToPlcEntity)
+                .collect(Collectors.toList());
+    }
+
+    /**
+     * 执行统计查询，返回总数
+     * @param flux Flux查询语句
+     * @return 统计结果
+     */
+    public long queryCount(String flux) {
+        List<FluxTable> tables = queryApi.query(flux);
+        if (tables.isEmpty()) {
+            return 0L;
+        }
+        
+        // 获取count结果
+        return tables.stream()
+                .flatMap(table -> table.getRecords().stream())
+                .mapToLong(record -> {
+                    Object value = record.getValue();
+                    if (value instanceof Number) {
+                        return ((Number) value).longValue();
+                    }
+                    return 0L;
+                })
+                .sum();
+    }
+
+    /**
+     * Flux结果转PLC实体
+     */
+    private PlcSignal convertToPlcEntity(FluxRecord record) {
+        return PlcSignal.builder()
+                .device(record.getValueByKey("device").toString())
+                .tag(record.getValueByKey("tag").toString())
+                .value(record.getValueByKey("value").toString())
+                .changeTime(record.getTime())
+                .build();
+    }
+}
--- a/nl-iot/src/main/java/org/nl/iot/modular/influxdb/entity/PageResult.java
+++ b/nl-iot/src/main/java/org/nl/iot/modular/influxdb/entity/PageResult.java
@@ -0,0 +1,112 @@
+package org.nl.iot.modular.influxdb.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+/**
+ * 分页结果封装类
+ * @param <T> 数据类型
+ * @Author: liyongde
+ * @Date: 2026/3/24
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class PageResult<T> {
+    
+    /**
+     * 当前页数据列表
+     */
+    private List<T> records;
+    
+    /**
+     * 总记录数
+     */
+    private long total;
+    
+    /**
+     * 当前页码（从1开始）
+     */
+    private int page;
+    
+    /**
+     * 每页大小
+     */
+    private int size;
+    
+    /**
+     * 总页数
+     */
+    private int totalPages;
+    
+    /**
+     * 是否有下一页
+     */
+    private boolean hasNext;
+    
+    /**
+     * 是否有上一页
+     */
+    private boolean hasPrevious;
+    
+    /**
+     * 是否为第一页
+     */
+    private boolean isFirst;
+    
+    /**
+     * 是否为最后一页
+     */
+    private boolean isLast;
+    
+    /**
+     * 构建分页结果
+     * @param records 当前页数据
+     * @param total 总记录数
+     * @param page 当前页码
+     * @param size 每页大小
+     * @param <T> 数据类型
+     * @return 分页结果
+     */
+    public static <T> PageResult<T> of(List<T> records, long total, int page, int size) {
+        int totalPages = (int) Math.ceil((double) total / size);
+        
+        return PageResult.<T>builder()
+                .records(records)
+                .total(total)
+                .page(page)
+                .size(size)
+                .totalPages(totalPages)
+                .hasNext(page < totalPages)
+                .hasPrevious(page > 1)
+                .isFirst(page == 1)
+                .isLast(page >= totalPages)
+                .build();
+    }
+    
+    /**
+     * 空分页结果
+     * @param page 当前页码
+     * @param size 每页大小
+     * @param <T> 数据类型
+     * @return 空分页结果
+     */
+    public static <T> PageResult<T> empty(int page, int size) {
+        return PageResult.<T>builder()
+                .records(List.of())
+                .total(0L)
+                .page(page)
+                .size(size)
+                .totalPages(0)
+                .hasNext(false)
+                .hasPrevious(page > 1)
+                .isFirst(page == 1)
+                .isLast(true)
+                .build();
+    }
+}
--- a/nl-iot/src/main/java/org/nl/iot/modular/influxdb/entity/PlcSignal.java
+++ b/nl-iot/src/main/java/org/nl/iot/modular/influxdb/entity/PlcSignal.java
@@ -0,0 +1,48 @@
+package org.nl.iot.modular.influxdb.entity;
+
+import com.influxdb.annotations.Column;
+import com.influxdb.annotations.Measurement;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+
+/**
+ * plc信号实体
+ * @Author: liyongde
+ * @Date: 2026/3/24 10:53
+ * @Modified By:
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@Measurement(name = "plc_signal")
+public class PlcSignal {
+
+    /**
+     * 设备名称（Tag索引）
+     */
+    @Column(tag = true, name = "device")
+    private String device;
+
+    /**
+     * 信号点位/标签（Tag索引）
+     */
+    @Column(tag = true, name = "tag")
+    private String tag;
+
+    /**
+     * 信号值（Field，存变化数据）
+     */
+    @Column(name = "value")
+    private String value;
+
+    /**
+     * 变化时间戳（InfluxDB自动维护）
+     */
+    @Column(timestamp = true)
+    private Instant changeTime;
+}
--- a/nl-iot/src/main/java/org/nl/iot/modular/influxdb/service/PlcSignalService.java
+++ b/nl-iot/src/main/java/org/nl/iot/modular/influxdb/service/PlcSignalService.java
@@ -0,0 +1,38 @@
+package org.nl.iot.modular.influxdb.service;
+
+import org.nl.iot.core.driver.entity.RValue;
+import org.springframework.scheduling.annotation.Async;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * influxdb存储信号变化值
+ * @Author: liyongde
+ * @Date: 2026/3/24 13:11
+ */
+public interface PlcSignalService {
+
+    /**
+     * 存储变化值
+     * @param device
+     * @param tag
+     * @param newValue
+     * @return
+     */
+    void savePlcChangeSignal(String device, String tag, String newValue);
+
+    /**
+     * 批量存储（同步）
+     * @param needSaveValues
+     */
+    void batchSaveSignalToInfluxDB(List<RValue> needSaveValues);
+
+    /**
+     * 批量存储（异步）
+     * @param needSaveValues
+     * @return CompletableFuture<Void>
+     */
+    @Async("influxDbAsyncExecutor")
+    CompletableFuture<Void> batchSaveSignalToInfluxDBAsync(List<RValue> needSaveValues);
+}
--- a/nl-iot/src/main/java/org/nl/iot/modular/influxdb/service/impl/PlcSignalServiceImpl.java
+++ b/nl-iot/src/main/java/org/nl/iot/modular/influxdb/service/impl/PlcSignalServiceImpl.java
@@ -0,0 +1,80 @@
+package org.nl.iot.modular.influxdb.service.impl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.nl.iot.core.driver.bo.SiteBO;
+import org.nl.iot.core.driver.entity.RValue;
+import org.nl.iot.core.influxDB.InfluxDB2Util;
+import org.nl.iot.modular.influxdb.entity.PlcSignal;
+import org.nl.iot.modular.influxdb.service.PlcSignalService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ *
+ * @Author: liyongde
+ * @Date: 2026/3/24 13:11
+ */
+@Slf4j
+@Service
+public class PlcSignalServiceImpl implements PlcSignalService {
+
+    @Autowired
+    private InfluxDB2Util influxDB2Util;
+
+    @Override
+    public void savePlcChangeSignal(String device, String tag, String newValue) {
+        PlcSignal plcSignal = PlcSignal.builder()
+                .device(device)
+                .tag(tag)
+                .value(newValue)
+                .changeTime(Instant.now())
+                .build();
+        influxDB2Util.insertOne(plcSignal);
+    }
+
+    @Override
+    public void batchSaveSignalToInfluxDB(List<RValue> needSaveValues) {
+        if (needSaveValues == null || needSaveValues.isEmpty()) {
+            return;
+        }
+        
+        List<PlcSignal> plcSignals = new ArrayList<>();
+        for (RValue needSaveValue : needSaveValues) {
+            SiteBO siteBO = needSaveValue.getSiteBO();
+            PlcSignal plcSignal = PlcSignal.builder()
+                    .device(siteBO.getDeviceCode())
+                    .tag(siteBO.getAlias())
+                    .value(needSaveValue.getValue())
+                    .changeTime(Instant.now())
+                    .build();
+            plcSignals.add(plcSignal);
+        }
+        
+        try {
+            influxDB2Util.insertBatch(plcSignals);
+            log.debug("批量保存信号到InfluxDB成功，数量: {}", plcSignals.size());
+        } catch (Exception e) {
+            log.error("批量保存信号到InfluxDB失败，数量: {}", plcSignals.size(), e);
+            throw e;
+        }
+    }
+
+    @Override
+    @Async("influxDbAsyncExecutor")
+    public CompletableFuture<Void> batchSaveSignalToInfluxDBAsync(List<RValue> needSaveValues) {
+        try {
+            batchSaveSignalToInfluxDB(needSaveValues);
+            log.debug("异步批量保存信号到InfluxDB成功，数量: {}", needSaveValues.size());
+            return CompletableFuture.completedFuture(null);
+        } catch (Exception e) {
+            log.error("异步批量保存信号到InfluxDB失败，数量: {}", needSaveValues.size(), e);
+            return CompletableFuture.failedFuture(e);
+        }
+    }
+}
--- a/nl-web-app/src/main/java/org/nl/Application.java
+++ b/nl-web-app/src/main/java/org/nl/Application.java
@@ -19,6 +19,7 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.core.env.Environment;
+import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
@@ -33,6 +34,7 @@ import org.springframework.web.bind.annotation.RestController;
@RestController
@SpringBootApplication
@EnableScheduling
+@EnableAsync
 public class Application {

    /* 解决druid 日志报错：discard long time none received connection:xxx */
--- a/nl-web-app/src/main/resources/application.yml
+++ b/nl-web-app/src/main/resources/application.yml
@@ -177,3 +177,13 @@ file:
  # 文件大小 /M
  maxSize: 100
  avatarMaxSize: 5
+# InfluxDB 2.x 核心配置
+influx:
+  # 服务地址
+  url: http://117.72.202.142:15964/
+  # 初始化生成的All-Access令牌
+  token: mr8DZXza-WvBtHhK51Jnfh0tNlq_rLyVYW4D9Zyz0IN66RM9KqgS8agV4XWtXekCgGhG866Dtb9Jv0t-N5D0Gw==
+  # 组织名称
+  org: acs
+  # 数据桶名称（替代1.x的database）
+  bucket: signals
--- a/nl-web-app/src/test/java/org/nl/AsyncInfluxDBTest.java
+++ b/nl-web-app/src/test/java/org/nl/AsyncInfluxDBTest.java
@@ -0,0 +1,116 @@
+package org.nl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.nl.iot.core.driver.bo.SiteBO;
+import org.nl.iot.core.driver.entity.RValue;
+import org.nl.iot.modular.influxdb.service.PlcSignalService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ *
+ * @Author: liyongde
+ * @Date: 2026/3/24 13:58
+ */
+@Slf4j
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = Application.class)
+public class AsyncInfluxDBTest {
+    @Autowired
+    private PlcSignalService plcSignalService;
+
+    /**
+     * 测试异步批量保存
+     */
+    @Test
+    public void testAsyncBatchSave() {
+        log.info("开始测试异步批量保存...");
+
+        // 构造测试数据
+        List<RValue> testValues = createTestData();
+
+        // 记录开始时间
+        long startTime = System.currentTimeMillis();
+
+        // 异步保存
+        CompletableFuture<Void> future = plcSignalService.batchSaveSignalToInfluxDBAsync(testValues);
+
+        // 主线程继续执行其他任务
+        log.info("异步任务已提交，主线程继续执行其他任务...");
+        doOtherWork();
+
+        // 等待异步任务完成（可选）
+        future.whenComplete((result, throwable) -> {
+            long duration = System.currentTimeMillis() - startTime;
+            if (throwable != null) {
+                log.error("异步保存失败，耗时: {}ms", duration, throwable);
+            } else {
+                log.info("异步保存成功，耗时: {}ms", duration);
+            }
+        });
+
+        log.info("测试方法执行完成");
+    }
+
+    /**
+     * 对比同步和异步性能
+     */
+    @Test
+    public void comparePerformance() {
+        List<RValue> testValues = createTestData();
+
+        // 测试同步方式
+        long syncStart = System.currentTimeMillis();
+        plcSignalService.batchSaveSignalToInfluxDB(testValues);
+        long syncDuration = System.currentTimeMillis() - syncStart;
+        log.info("同步保存耗时: {}ms", syncDuration);
+
+        // 测试异步方式
+        long asyncStart = System.currentTimeMillis();
+        CompletableFuture<Void> future = plcSignalService.batchSaveSignalToInfluxDBAsync(testValues);
+        long asyncSubmitDuration = System.currentTimeMillis() - asyncStart;
+        log.info("异步任务提交耗时: {}ms", asyncSubmitDuration);
+
+        // 等待异步任务完成
+        future.join();
+        long asyncTotalDuration = System.currentTimeMillis() - asyncStart;
+        log.info("异步任务总耗时: {}ms", asyncTotalDuration);
+    }
+
+    private List<RValue> createTestData() {
+        List<RValue> testValues = new ArrayList<>();
+
+        for (int i = 0; i < 100; i++) {
+            SiteBO siteBO = SiteBO.builder()
+                    .deviceCode("TEST_DEVICE_" + (i % 10))
+                    .alias("TEST_TAG_" + i)
+                    .build();
+
+            RValue rValue = RValue.builder()
+                    .siteBO(siteBO)
+                    .value("TEST_VALUE_" + i)
+                    .build();
+
+            testValues.add(rValue);
+        }
+
+        return testValues;
+    }
+
+    private void doOtherWork() {
+        // 模拟其他业务逻辑
+        try {
+            Thread.sleep(100);
+            log.info("其他业务逻辑执行完成");
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}
--- a/nl-web-app/src/test/resources/application-test.yml
+++ b/nl-web-app/src/test/resources/application-test.yml
@@ -0,0 +1,76 @@
+# 测试环境配置
+spring:
+  profiles:
+    active: test
+  main:
+    allow-circular-references: true
+  datasource:
+    dynamic:
+      datasource:
+        master:
+          driver-class-name: com.mysql.cj.jdbc.Driver
+          url: jdbc:mysql://192.168.81.251:3306/acs3.0?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true&useInformationSchema=true&rewriteBatchedStatements=true
+          username: root
+          password: "P@ssw0rd."
+      strict: true
+      public-key: MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBAMWiTVtdXFVrgFHDDKELZM0SywkWY3KjugN90eY5Sogon1j8Y0ClPF7nx3FuE7pAeBKiv7ChIS0vvx/59WUpKmUCAwEAAQ==
+  data:
+    redis:
+      database: 1
+      host: 127.0.0.1
+      port: 6379
+      password: ""
+      timeout: 10s
+  jackson:
+    time-zone: GMT+8
+    date-format: "yyyy-MM-dd HH:mm:ss"
+    locale: zh_CN
+    serialization:
+      write-dates-as-timestamps: false
+
+# InfluxDB 2.x 测试配置
+influx:
+  # 服务地址
+  url: http://117.72.202.142:15964/
+  # 初始化生成的All-Access令牌
+  token: mr8DZXza-WvBtHhK51Jnfh0tNlq_rLyVYW4D9Zyz0IN66RM9KqgS8agV4XWtXekCgGhG866Dtb9Jv0t-N5D0Gw==
+  # 组织名称
+  org: acs
+  # 数据桶名称（替代1.x的database）
+  bucket: signals
+
+mybatis-plus:
+  configuration:
+    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
+    jdbc-type-for-null: "null"
+  global-config:
+    banner: false
+    enable-sql-runner: true
+    db-config:
+      id-type: ASSIGN_ID
+      logic-delete-field: DELETE_FLAG
+      logic-delete-value: DELETED
+      logic-not-delete-value: NOT_DELETE
+  mapper-locations: classpath*:org/nl/**/mapping/*.xml
+  type-handlers-package: org.nl.common.handler
+
+# sa-token configuration
+sa-token:
+  token-name: token
+  timeout: 2592000
+  active-timeout: -1
+  is-concurrent: true
+  is-share: false
+  max-login-count: -1
+  token-style: random-32
+  is-log: false
+  is-print: false
+
+#########################################
+# easy-trans configuration
+#########################################
+easy-trans:
+  is-enable-redis: true
+  is-enable-global: true
+  is-enable-tile: true
+  is-enable-cloud: false
--- a/pom.xml
+++ b/pom.xml
@@ -451,6 +451,47 @@
                <artifactId>mssql-jdbc</artifactId>
                <version>12.4.2.jre11</version>
            </dependency>-->
+
+            <!-- OPC UA 相关 -->
+            <dependency>
+                <groupId>org.eclipse.milo</groupId>
+                <artifactId>sdk-client</artifactId>
+                <version>0.6.16</version>
+            </dependency>
+
+            <!-- Java COM opc da 相关 -->
+            <dependency>
+                <groupId>org.jinterop</groupId>
+                <artifactId>j-interop</artifactId>
+                <version>2.0.4</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.plc4x</groupId>
+                <artifactId>plc4j-api</artifactId>
+                <version>0.13.0</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.plc4x</groupId>
+                <artifactId>plc4j-driver-s7</artifactId>
+                <version>0.13.0</version>
+                <scope>runtime</scope>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.plc4x</groupId>
+                <artifactId>plc4j-driver-modbus</artifactId>
+                <version>0.13.0</version>
+                <scope>runtime</scope>
+            </dependency>
+
+            <!-- InfluxDB 2.x 官方Java客户端 -->
+            <dependency>
+                <groupId>com.influxdb</groupId>
+                <artifactId>influxdb-client-java</artifactId>
+                <version>6.11.0</version>
+            </dependency>
        </dependencies>
    </dependencyManagement>
Author	SHA1	Message	Date
liyongde	de0c031b2b	fix: influxdb数据库	2026-03-24 14:47:52 +08:00
liyongde	ae95d4dfa3	feat: 新增influxdb数据库	2026-03-24 13:50:50 +08:00