package com.zhitan.realtimedata.data.influxdb;
|
|
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.influxdb.query.FluxRecord;
|
import com.influxdb.query.FluxTable;
|
import com.zhitan.common.enums.CollectionModes;
|
import com.zhitan.common.enums.GroupTimeType;
|
import com.zhitan.common.enums.Quality;
|
import com.zhitan.realtimedata.config.RtdbConfig;
|
import com.zhitan.realtimedata.domain.TagValue;
|
import com.zhitan.realtimedata.domain.WritePoint;
|
import lombok.extern.slf4j.Slf4j;
|
import org.joda.time.DateTime;
|
import org.joda.time.Duration;
|
import org.springframework.stereotype.Repository;
|
|
import java.time.Instant;
|
import java.util.ArrayList;
|
import java.util.Collections;
|
import java.util.Date;
|
import java.util.List;
|
|
/**
|
* 实时库初连接
|
*
|
* @Author: Zhujw
|
* @Date: 2023/5/24
|
*/
|
@Repository
|
@Slf4j
|
public class InfluxDBRepository {
|
|
private final RtdbConfig config;
|
|
private InfluxDBClient client;
|
|
public InfluxDBRepository(RtdbConfig 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, Date beginTime, Date endTime) {
|
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);
|
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);
|
}
|
|
public List<TagValue> getHistoryData(List<String> tagCodes, Date beginTime, Date endTime,
|
int pointCount) {
|
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);
|
}
|
}
|
