Compare commits
2 Commits
1ff30fd30d
...
de0c031b2b
| Author | SHA1 | Date | |
|---|---|---|---|
| de0c031b2b | |||
| ae95d4dfa3 |
@@ -28,35 +28,36 @@
|
|||||||
<dependency>
|
<dependency>
|
||||||
<groupId>org.eclipse.milo</groupId>
|
<groupId>org.eclipse.milo</groupId>
|
||||||
<artifactId>sdk-client</artifactId>
|
<artifactId>sdk-client</artifactId>
|
||||||
<version>0.6.16</version>
|
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
<!-- Java COM opc da 相关 -->
|
<!-- Java COM opc da 相关 -->
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>org.jinterop</groupId>
|
<groupId>org.jinterop</groupId>
|
||||||
<artifactId>j-interop</artifactId>
|
<artifactId>j-interop</artifactId>
|
||||||
<version>2.0.4</version>
|
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>org.apache.plc4x</groupId>
|
<groupId>org.apache.plc4x</groupId>
|
||||||
<artifactId>plc4j-api</artifactId>
|
<artifactId>plc4j-api</artifactId>
|
||||||
<version>0.13.0</version>
|
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>org.apache.plc4x</groupId>
|
<groupId>org.apache.plc4x</groupId>
|
||||||
<artifactId>plc4j-driver-s7</artifactId>
|
<artifactId>plc4j-driver-s7</artifactId>
|
||||||
<version>0.13.0</version>
|
|
||||||
<scope>runtime</scope>
|
<scope>runtime</scope>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>org.apache.plc4x</groupId>
|
<groupId>org.apache.plc4x</groupId>
|
||||||
<artifactId>plc4j-driver-modbus</artifactId>
|
<artifactId>plc4j-driver-modbus</artifactId>
|
||||||
<version>0.13.0</version>
|
|
||||||
<scope>runtime</scope>
|
<scope>runtime</scope>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
|
<!-- InfluxDB 2.x 官方Java客户端 -->
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.influxdb</groupId>
|
||||||
|
<artifactId>influxdb-client-java</artifactId>
|
||||||
|
</dependency>
|
||||||
</dependencies>
|
</dependencies>
|
||||||
|
|
||||||
</project>
|
</project>
|
||||||
56
nl-iot/src/main/java/org/nl/iot/config/AsyncConfig.java
Normal file
56
nl-iot/src/main/java/org/nl/iot/config/AsyncConfig.java
Normal file
@@ -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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,11 +1,13 @@
|
|||||||
package org.nl.iot.core.cache;
|
package org.nl.iot.core.cache;
|
||||||
|
|
||||||
|
import cn.hutool.core.util.ObjectUtil;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.nl.iot.core.driver.bo.ConnectSiteDO;
|
import org.nl.iot.core.driver.bo.ConnectSiteDO;
|
||||||
import org.nl.iot.core.driver.bo.DeviceConnectionBO;
|
import org.nl.iot.core.driver.bo.DeviceConnectionBO;
|
||||||
import org.nl.iot.core.driver.bo.DeviceInfoReadCacheDo;
|
import org.nl.iot.core.driver.bo.DeviceInfoReadCacheDo;
|
||||||
import org.nl.iot.core.driver.bo.SiteBO;
|
import org.nl.iot.core.driver.bo.SiteBO;
|
||||||
import org.nl.iot.core.driver.entity.RValue;
|
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.nl.iot.modular.iot.service.IotConfigService;
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
import org.springframework.boot.CommandLineRunner;
|
import org.springframework.boot.CommandLineRunner;
|
||||||
@@ -29,6 +31,9 @@ public class MetadataCacheManager implements CommandLineRunner {
|
|||||||
@Autowired
|
@Autowired
|
||||||
private IotConfigService configService;
|
private IotConfigService configService;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private PlcSignalService plcSignalService;
|
||||||
|
|
||||||
private final Map<String, DeviceInfoReadCacheDo> deviceInfoCache = new ConcurrentHashMap<>();
|
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
|
* @param deviceKey connectCode.deviceCode
|
||||||
@@ -203,12 +216,27 @@ public class MetadataCacheManager implements CommandLineRunner {
|
|||||||
if (rValues == null || rValues.isEmpty()) {
|
if (rValues == null || rValues.isEmpty()) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
Map<String, RValue> cacheSnapshot = getAllSignalValues();
|
||||||
for (RValue rValue : rValues) {
|
List<RValue> needSaveValues = new ArrayList<>();
|
||||||
if (rValue != null && rValue.getSiteBO() != null) {
|
for (RValue newRValue : rValues) {
|
||||||
String alias = rValue.getSiteBO().getAlias();
|
if (newRValue != null && newRValue.getSiteBO() != null) {
|
||||||
putSignalValue(deviceKey, alias, rValue);
|
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;
|
||||||
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
105
nl-iot/src/main/java/org/nl/iot/core/influxDB/InfluxDB2Util.java
Normal file
105
nl-iot/src/main/java/org/nl/iot/core/influxDB/InfluxDB2Util.java
Normal file
@@ -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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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;
|
||||||
|
}
|
||||||
@@ -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);
|
||||||
|
}
|
||||||
@@ -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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -19,6 +19,7 @@ import org.springframework.boot.SpringApplication;
|
|||||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||||
import org.springframework.context.ConfigurableApplicationContext;
|
import org.springframework.context.ConfigurableApplicationContext;
|
||||||
import org.springframework.core.env.Environment;
|
import org.springframework.core.env.Environment;
|
||||||
|
import org.springframework.scheduling.annotation.EnableAsync;
|
||||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||||
import org.springframework.web.bind.annotation.GetMapping;
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
import org.springframework.web.bind.annotation.RestController;
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
@@ -33,6 +34,7 @@ import org.springframework.web.bind.annotation.RestController;
|
|||||||
@RestController
|
@RestController
|
||||||
@SpringBootApplication
|
@SpringBootApplication
|
||||||
@EnableScheduling
|
@EnableScheduling
|
||||||
|
@EnableAsync
|
||||||
public class Application {
|
public class Application {
|
||||||
|
|
||||||
/* 解决druid 日志报错:discard long time none received connection:xxx */
|
/* 解决druid 日志报错:discard long time none received connection:xxx */
|
||||||
|
|||||||
@@ -177,3 +177,13 @@ file:
|
|||||||
# 文件大小 /M
|
# 文件大小 /M
|
||||||
maxSize: 100
|
maxSize: 100
|
||||||
avatarMaxSize: 5
|
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
|
||||||
116
nl-web-app/src/test/java/org/nl/AsyncInfluxDBTest.java
Normal file
116
nl-web-app/src/test/java/org/nl/AsyncInfluxDBTest.java
Normal file
@@ -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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
76
nl-web-app/src/test/resources/application-test.yml
Normal file
76
nl-web-app/src/test/resources/application-test.yml
Normal file
@@ -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
|
||||||
41
pom.xml
41
pom.xml
@@ -451,6 +451,47 @@
|
|||||||
<artifactId>mssql-jdbc</artifactId>
|
<artifactId>mssql-jdbc</artifactId>
|
||||||
<version>12.4.2.jre11</version>
|
<version>12.4.2.jre11</version>
|
||||||
</dependency>-->
|
</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>
|
</dependencies>
|
||||||
</dependencyManagement>
|
</dependencyManagement>
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user