package com.zhitan.engine.influxdb;
|
|
import com.influxdb.LogLevel;
|
import com.influxdb.client.InfluxDBClient;
|
import com.influxdb.client.InfluxDBClientFactory;
|
import com.influxdb.client.QueryApi;
|
import com.influxdb.client.WriteApiBlocking;
|
import com.influxdb.client.domain.HealthCheck;
|
import com.influxdb.client.domain.WritePrecision;
|
import com.zhitan.engine.enums.CollectionModes;
|
import com.zhitan.engine.enums.GroupTimeType;
|
import com.zhitan.engine.enums.Quality;
|
import com.influxdb.query.FluxRecord;
|
import com.influxdb.query.FluxTable;
|
import com.zhitan.engine.config.InfluxDBConfig;
|
import com.zhitan.engine.entity.TagValue;
|
import com.zhitan.engine.entity.WritePoint;
|
import lombok.extern.slf4j.Slf4j;
|
import org.joda.time.DateTime;
|
import org.joda.time.Duration;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.stereotype.Repository;
|
|
import java.time.*;
|
import java.time.format.DateTimeFormatter;
|
import java.util.ArrayList;
|
import java.util.Collections;
|
import java.util.Date;
|
import java.util.List;
|
|
/**
|
* influxdb的基础服务
|
*
|
* @author Silence
|
* @version 1.0
|
*/
|
@Slf4j
|
@Repository
|
public class InfluxdbRepository {
|
|
protected InfluxDBConfig config;
|
protected InfluxDBClient client;
|
|
@Value("${influxdb.bucket}")
|
private String database;
|
|
public InfluxdbRepository(InfluxDBConfig config) {
|
this.config = config;
|
connectInfluxDB();
|
}
|
|
private void connectInfluxDB() {
|
char[] token = config.getToken().toCharArray();
|
if (client == null) {
|
client = InfluxDBClientFactory.create(config.getHost(), token, config.getOrg(),
|
config.getBucket());
|
}
|
log.error("--------------------实时数据库连接成功--------------------");
|
HealthCheck health = client.health();
|
if (health.getStatus() == HealthCheck.StatusEnum.FAIL) {
|
client.close();
|
client = InfluxDBClientFactory.create(config.getHost(), token, config.getOrg(),
|
config.getBucket());
|
}
|
}
|
|
public void store(List<TagValue> tagValues) {
|
WriteApiBlocking write = client.getWriteApiBlocking();
|
List<WritePoint> writePoints = new ArrayList<>();
|
tagValues.forEach(tagValue -> {
|
WritePoint point = new WritePoint(tagValue.getTagCode(), tagValue.getValue(),
|
Instant.ofEpochMilli(tagValue.getDataTime().getTime()));
|
writePoints.add(point);
|
});
|
write.writeMeasurements(WritePrecision.MS, writePoints);
|
}
|
|
public TagValue query(String tagCode, Date time) {
|
List<TagValue> values = query(Collections.singletonList(tagCode), time);
|
return !values.isEmpty() ? values.get(0) : new TagValue();
|
}
|
|
public TagValue query(String tagCode) {
|
List<TagValue> values = query(Collections.singletonList(tagCode));
|
return !values.isEmpty() ? values.get(0) : new TagValue();
|
}
|
|
public List<TagValue> query(List<String> tagCodes, Date time) {
|
DateTime endTime = new DateTime(time);
|
DateTime beginTime = endTime.plusDays(-1);
|
return getTagValues(beginTime, endTime, tagCodes);
|
}
|
|
public List<TagValue> query(List<String> tagCodes) {
|
DateTime beginTime = DateTime.now().plusDays(-1);
|
DateTime endTime = DateTime.now();
|
return getTagValues(beginTime, endTime, tagCodes);
|
}
|
//
|
// /**
|
// * 获取一个小时内所有数据
|
// *
|
// * @param tagCodes 测点编号
|
// * @param time 开始时间
|
// * @return 测点统计结果
|
// */
|
// public List<TagValue> queryOneHour(List<String> tagCodes, Date time) {
|
// DateTime beginTime = new DateTime(time);
|
// DateTime endTime = beginTime.plusHours(1);
|
// return getHistoryData(tagCodes, beginTime.toDate(), endTime.toDate());
|
// }
|
|
private List<TagValue> getHistoryData(List<String> tagCodes, LocalDateTime beginTime, LocalDateTime endTime) {
|
// 将 beginTime 转换为 java.util.Date 对象
|
Date o = Date.from(beginTime.atZone(ZoneId.systemDefault()).toInstant());
|
DateTime begin = new DateTime(o);
|
Date o1 = Date.from(endTime.atZone(ZoneId.systemDefault()).toInstant());
|
DateTime end = new DateTime(o1);
|
StringBuilder timeRange = new StringBuilder()
|
.append("|> range(start: ").append(Instant.ofEpochMilli(begin.getMillis()).toString())
|
.append(", stop: ").append(Instant.ofEpochMilli(end.getMillis()).toString())
|
.append(")");
|
StringBuilder fluxSql = getStringBuilder(timeRange,
|
tagCodes);
|
log.info(String.valueOf(fluxSql));
|
return getTagValues(fluxSql);
|
}
|
|
public List<TagValue> getHistoryData(List<String> tagCodes, Date beginTime, Date endTime,
|
long interval) {
|
DateTime begin = new DateTime(beginTime);
|
DateTime end = new DateTime(endTime);
|
StringBuilder timeRange = new StringBuilder()
|
.append("|> range(start: ").append(Instant.ofEpochMilli(begin.getMillis()).toString())
|
.append(", stop: ").append(Instant.ofEpochMilli(end.getMillis()).toString())
|
.append(")");
|
StringBuilder fluxSql = getStringBuilder(timeRange, tagCodes);
|
fluxSql.append("|> aggregateWindow(every: ").append(interval)
|
.append("s, fn: last, createEmpty: false)");
|
log.info(String.valueOf(fluxSql));
|
return getTagValues(fluxSql);
|
}
|
|
/**
|
* 获取指定标签代码在时间范围内的历史数据，并按指定点数计算数据间隔
|
*
|
* @param tagCodes 标签代码列表，用于标识需要获取历史数据的设备/变量
|
* @param beginTime 数据查询的起始时间（包含）
|
* @param endTime 数据查询的结束时间（包含）
|
* @param pointCount 期望获取的数据点数，用于自动计算采样间隔。实际点数可能受存储限制影响
|
* @return List<TagValue> 包含时间序列数据的对象集合，按时间顺序排列
|
*
|
* 方法逻辑：
|
* 1. 将Date类型参数转换为Joda-Time的DateTime类型以便计算时间间隔
|
* 2. 计算总时间跨度（毫秒），根据点数要求转换为秒级间隔
|
* 3. 调用底层方法获取按计算间隔采样的历史数据
|
*/
|
public List<TagValue> getHistoryData(List<String> tagCodes, Date beginTime, Date endTime,
|
int pointCount) {
|
// 转换时间对象为Joda-Time类型（兼容旧Date类型参数）
|
DateTime begin = new DateTime(beginTime);
|
DateTime end = new DateTime(endTime);
|
|
// 计算时间总跨度并转换为秒级采样间隔
|
long millis = new Duration(begin, end).getMillis();
|
long interval = millis / pointCount / 1000;
|
|
// 委托给核心方法执行实际数据查询
|
return getHistoryData(tagCodes, beginTime, endTime, interval);
|
}
|
|
|
private StringBuilder getStringBuilder(StringBuilder timeRange,
|
List<String> tagCodes) {
|
StringBuilder fluxSql = new StringBuilder();
|
fluxSql.append("from(bucket: \"").append(config.getBucket()).append("\")")
|
.append(timeRange).append("|> filter(fn: (r) => r[\"_measurement\"] == \"")
|
.append(config.getMeasurement()).append("\")");
|
fluxSql.append("|> filter(fn: (r) => r[\"_field\"] == \"value\")");
|
if (!tagCodes.isEmpty()) {
|
fluxSql.append("|> filter(fn: (r) => r[\"tag\"] =~ /");
|
List<String> filter = new ArrayList<>(tagCodes);
|
fluxSql.append(String.join("|", filter));
|
fluxSql.append("/)");
|
}
|
fluxSql.append("|> group(columns: [\"tag\"])");
|
log.info(String.valueOf(fluxSql));
|
return fluxSql;
|
}
|
|
private List<TagValue> getTagValues(StringBuilder fluxSql) {
|
QueryApi query = client.getQueryApi();
|
|
List<FluxTable> tables = query.query(fluxSql.toString());
|
|
List<TagValue> values = new ArrayList<>();
|
for (FluxTable fluxTable : tables) {
|
List<FluxRecord> records = fluxTable.getRecords();
|
for (FluxRecord fluxRecord : records) {
|
String tag = String.valueOf(fluxRecord.getValueByKey("tag"));
|
Double value = (Double) fluxRecord.getValueByKey("_value");
|
Date time = fluxRecord.getTime() == null ? null
|
: new DateTime(fluxRecord.getTime().toString()).toDate();
|
TagValue tagValue = new TagValue();
|
tagValue.setValue(value);
|
tagValue.setDataTime(time);
|
tagValue.setTagCode(tag);
|
tagValue.setQuality(Quality.GOOD);
|
values.add(tagValue);
|
}
|
}
|
return values;
|
}
|
|
public List<TagValue> statistics(List<String> tagCodes, Date beginTime, Date endTime,
|
CollectionModes queryType) {
|
DateTime begin = new DateTime(beginTime);
|
DateTime end = new DateTime(endTime);
|
return getTagValues(begin, end, tagCodes, queryType);
|
}
|
|
public List<TagValue> statistics(List<String> tagCodes, Date beginTime, Date endTime,
|
CollectionModes queryType, GroupTimeType timeType) {
|
DateTime begin = new DateTime(beginTime);
|
DateTime end = new DateTime(endTime);
|
StringBuilder timeRange = new StringBuilder()
|
.append("|> range(start: ").append(Instant.ofEpochMilli(begin.getMillis()).toString())
|
.append(", stop: ").append(Instant.ofEpochMilli(end.getMillis()).toString())
|
.append(")");
|
StringBuilder fluxSql = getStringBuilder(timeRange, tagCodes);
|
fluxSql.append("|> aggregateWindow(every: 1")
|
.append(timeType.name())
|
.append(", fn: ")
|
.append(queryType.name())
|
.append(", createEmpty: false)")
|
.append("|>yield(name: \"")
|
.append(queryType.name())
|
.append("\")");
|
log.info(String.valueOf(fluxSql));
|
return getTagValues(fluxSql);
|
}
|
|
private List<TagValue> getTagValues(DateTime begin, DateTime end, List<String> tagCodes) {
|
return getTagValues(begin, end, tagCodes, CollectionModes.last);
|
}
|
|
private List<TagValue> getTagValues(DateTime begin, DateTime end, List<String> tagCodes,
|
CollectionModes queryType) {
|
StringBuilder timeRange = new StringBuilder()
|
.append("|> range(start: ").append(Instant.ofEpochMilli(begin.getMillis()).toString())
|
.append(", stop: ").append(Instant.ofEpochMilli(end.getMillis()).toString())
|
.append(")");
|
StringBuilder fluxSql = getStringBuilder(timeRange, tagCodes);
|
fluxSql.append("|> ");
|
fluxSql.append(queryType.name());
|
fluxSql.append("()");
|
log.info(String.valueOf(fluxSql));
|
return getTagValues(fluxSql);
|
}
|
|
/**
|
* 解析查询结果中的数值
|
*
|
* @param tables 查询结果
|
* @return 数值列表
|
*/
|
public List<Double> getFieldValues(List<TagValue> tables) {
|
if (tables == null || tables.isEmpty()) {
|
return null;
|
}
|
|
List<Double> values = new ArrayList<>();
|
for (TagValue tagValue : tables) {
|
|
Double value = tagValue.getValue();
|
if (value != null) {
|
values.add(value);
|
}
|
|
}
|
|
return values.isEmpty() ? null : values;
|
}
|
|
/**
|
* 计算指定时间范围内的电能累计值
|
*
|
* @param field 字段 00601925032100026643_5_ActiveZT
|
* @param startTime 开始时间
|
* @param endTime 结束时间
|
* @return 累计值
|
*/
|
public Double calculateAccumulatedValue(String field, LocalDateTime startTime, LocalDateTime endTime) {
|
try {
|
List<String> tagCodes = new ArrayList<>();
|
tagCodes.add(field);
|
List<TagValue> tables = getHistoryData(tagCodes, startTime, endTime);
|
List<Double> values = getFieldValues(tables);
|
|
if (values == null || values.isEmpty()) {
|
log.warn("未查询到数据，返回默认值0.0，measurement={}, tag={},startTime={}, endTime={}",
|
config.getMeasurement(), field, startTime, endTime);
|
return 0.0;
|
}
|
|
// 计算累计值（最后一个值减去第一个值）
|
Object firstValue = values.get(0);
|
Object lastValue = values.get(values.size() - 1);
|
|
if (firstValue instanceof Number && lastValue instanceof Number) {
|
double result = ((Number) lastValue).doubleValue() - ((Number) firstValue).doubleValue();
|
log.info("计算累计值成功：measurement={}, tag={}, 起始值={}, 结束值={}, 累计值={}",
|
config.getMeasurement(), field, firstValue, lastValue, result);
|
return result;
|
} else {
|
log.warn("数据类型不是数值类型，返回默认值0.0，firstValue={}, lastValue={}", firstValue, lastValue);
|
return 0.0;
|
}
|
} catch (Exception e) {
|
log.error("计算累计值异常，返回默认值0.0，measurement={}, tag={}, error={}",
|
config.getMeasurement(), field, e.getMessage(), e);
|
return 0.0;
|
}
|
}
|
}
|
