package com.dingzhuo.energy.dataservice.data.influxdb;
|
|
import com.google.common.collect.Lists;
|
import com.google.common.util.concurrent.Futures;
|
import com.google.common.util.concurrent.ListenableFuture;
|
import com.google.common.util.concurrent.ListeningExecutorService;
|
import com.google.common.util.concurrent.MoreExecutors;
|
import com.google.common.util.concurrent.ThreadFactoryBuilder;
|
|
import java.time.Instant;
|
import java.util.ArrayList;
|
import java.util.Date;
|
import java.util.List;
|
import java.util.concurrent.ExecutionException;
|
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.LinkedBlockingQueue;
|
import java.util.concurrent.ThreadFactory;
|
import java.util.concurrent.ThreadPoolExecutor;
|
import java.util.concurrent.TimeUnit;
|
|
import okhttp3.OkHttpClient;
|
import org.apache.logging.log4j.LogManager;
|
import org.apache.logging.log4j.Logger;
|
import org.influxdb.BatchOptions;
|
import org.influxdb.InfluxDB;
|
import org.influxdb.InfluxDBFactory;
|
import org.influxdb.dto.BatchPoints;
|
import org.influxdb.dto.Point;
|
import org.influxdb.dto.Query;
|
import org.influxdb.dto.QueryResult;
|
import org.influxdb.impl.InfluxDBMapper;
|
import org.joda.time.DateTime;
|
import org.joda.time.Duration;
|
|
public class FastDB {
|
|
private Logger logger = LogManager.getLogger(FastDB.class);
|
private InfluxDB influxDB;
|
private static final String TABLE_NAME = "daq";
|
|
private String dataBaseName;
|
private String host;
|
private int port;
|
|
public FastDB(String host, int port, String dataBaseName) {
|
this.dataBaseName = dataBaseName;
|
this.host = host;
|
this.port = port;
|
try {
|
this.open();
|
} catch (Exception e) {
|
logger.fatal(e);
|
}
|
}
|
|
private static InfluxDBMapper resultMapper;
|
|
/**
|
* 打开连接
|
*/
|
private void open() {
|
try {
|
OkHttpClient.Builder builder = new OkHttpClient.Builder()
|
.connectTimeout(30, TimeUnit.SECONDS)
|
.readTimeout(60, TimeUnit.SECONDS)
|
.retryOnConnectionFailure(true);
|
influxDB = InfluxDBFactory.connect("http://" + this.host + ":" + this.port, builder);
|
influxDB
|
.enableBatch(BatchOptions.DEFAULTS.jitterDuration(500).actions(2000).flushDuration(100));
|
influxDB.enableGzip();
|
influxDB.ping();
|
resultMapper = new InfluxDBMapper(influxDB);
|
logger.info("实时库连接成功！");
|
} catch (Exception ex) {
|
logger.fatal("实时库连接失败！" + ex);
|
}
|
}
|
|
/**
|
* 关闭连接
|
*/
|
public void close() {
|
influxDB.close();
|
}
|
|
/**
|
* 根据点位号获取实时数据
|
*
|
* @param tagCodes 点位号列表
|
* @returnS
|
*/
|
public List<RtdbResult> snapShot(List<String> tagCodes) {
|
final String sql = "SELECT LAST(value) as value, quality FROM %s WHERE %s and " +
|
"time<=(now()) and time>=(now()-20m) GROUP BY code";
|
List<List<String>> newList = Lists.partition(tagCodes, 10);
|
List<ListenableFuture<QueryResult>> taskList = new ArrayList<>();
|
ListeningExecutorService service =
|
MoreExecutors.listeningDecorator(customerThreadPool());
|
for (List<String> tags : newList) {
|
ListenableFuture<QueryResult> futureTask = service.submit(() -> {
|
StringBuilder sb = new StringBuilder();
|
String codes = formatCode(tags);
|
sb.append(String.format(sql,
|
TABLE_NAME, codes));
|
return builderResultValue(sb.toString());
|
});
|
taskList.add(futureTask);
|
}
|
List<RtdbResult> results = buildResult(taskList);
|
service.shutdown();
|
return results;
|
}
|
|
private ExecutorService customerThreadPool() {
|
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
|
.setNameFormat("fast-db-thread-pool-%d").build();
|
return new ThreadPoolExecutor(10, 50,
|
30, TimeUnit.SECONDS,
|
new LinkedBlockingQueue<>(1024), namedThreadFactory,
|
new ThreadPoolExecutor.AbortPolicy());
|
}
|
|
/**
|
* 根据点位号获取某一时刻的历史数据
|
*
|
* @param tagCodes 点位号集合
|
* @param dataTime 历史时刻
|
* @param timeCode 区分时间类型的time code
|
* @return List<RtdbResult>
|
*/
|
public List<RtdbResult> getDataByTime(List<String> tagCodes, Date dataTime, String timeCode) {
|
DateTime begin = new DateTime(dataTime).plusDays(-1);;
|
// if (timeCode.startsWith("M")) {
|
// begin = new DateTime(dataTime).plusMonths(-1);
|
// } else if (timeCode.startsWith("Y")) {
|
// begin = new DateTime(dataTime).plusYears(-1);
|
// } else {
|
// begin = new DateTime(dataTime).plusDays(-1);
|
// }
|
DateTime end = new DateTime(dataTime);
|
// logger.error("开始计算数据，timeCode=" + timeCode + "开始时间=" + begin.toString("yyyy-MM-dd HH:mm:ss") + "结束时间=" + end.toString("yyyy-MM-dd HH:mm:ss"));
|
final String sql =
|
"SELECT LAST(value) as value, quality FROM %s WHERE %s AND time >=%dms AND " +
|
"time<=%dms GROUP BY code";
|
|
List<List<String>> newList = Lists.partition(tagCodes, 10);
|
List<ListenableFuture<QueryResult>> taskList = new ArrayList<>();
|
ListeningExecutorService service =
|
MoreExecutors.listeningDecorator(customerThreadPool());
|
for (List<String> tags : newList) {
|
ListenableFuture<QueryResult> futureTask = service.submit(() -> {
|
StringBuilder sb = new StringBuilder();
|
String codes = formatCode(tags);
|
sb.append(String.format(sql,
|
TABLE_NAME, codes, begin.toDate().getTime(), end.toDate().getTime() + 999));
|
return builderResultValue(sb.toString());
|
});
|
taskList.add(futureTask);
|
}
|
List<RtdbResult> results = buildResult(taskList);
|
service.shutdown();
|
return results;
|
}
|
|
/**
|
* 根据查询方式获取一段时间内的历史数据
|
*
|
* @param tagCodes 点位号集合
|
* @param beginTime 开始时间
|
* @param endTime 结束时间
|
* @param queryType 查询方式（循环、拟合、全部）
|
* @param pointCount 要查询的数据个数
|
* @return
|
*/
|
public List<RtdbResult> getHistoryData(List<String> tagCodes, Date beginTime, Date endTime,
|
QueryType queryType, int pointCount) {
|
DateTime begin = new DateTime(beginTime);
|
DateTime end = new DateTime(endTime);
|
long millis = new Duration(begin, end).getMillis();
|
List<List<String>> newList = Lists.partition(tagCodes, 10);
|
List<ListenableFuture<QueryResult>> taskList = new ArrayList<>();
|
|
ListeningExecutorService service =
|
MoreExecutors.listeningDecorator(customerThreadPool());
|
|
final String sqlFull =
|
"select value,quality FROM %s WHERE %s AND time >=%dms AND time <=%dms " +
|
"GROUP BY code";
|
final String sql = "select %s(value) as value, quality FROM %s WHERE %s AND time >=%dms " +
|
"AND" +
|
" time <=%dms GROUP BY code,time(%dms) fill(previous)";
|
for (List<String> tags : newList) {
|
ListenableFuture<QueryResult> futureTask = service.submit(() -> {
|
StringBuilder sb = new StringBuilder();
|
String codes = formatCode(tags);
|
if (queryType != QueryType.FULL && pointCount > 0) {
|
long interval = millis / pointCount;
|
sb.append(String.format(sql, queryType.name(),
|
TABLE_NAME, codes, begin.toDate().getTime(), end.toDate().getTime() + 999, interval));
|
} else {
|
sb.append(String.format(sqlFull,
|
TABLE_NAME, codes, begin.toDate().getTime(), end.toDate().getTime() + 999));
|
}
|
|
return builderResultValue(sb.toString());
|
});
|
taskList.add(futureTask);
|
}
|
List<RtdbResult> results = buildResult(taskList);
|
service.shutdown();
|
return results;
|
}
|
|
/**
|
* 汇总实时数据
|
*
|
* @param tagCodes 点位号集合
|
* @param beginTime 开始时间
|
* @param endTime 结束时间
|
* @param queryType 汇总方式（最大值、最小值、平均值、求和等）
|
* @return
|
*/
|
public List<RtdbResult> statistics(List<String> tagCodes, Date beginTime, Date endTime,
|
QueryType queryType) throws ExecutionException, InterruptedException {
|
if (queryType == QueryType.FULL) {
|
return new ArrayList<>();
|
}
|
DateTime begin = new DateTime(beginTime);
|
DateTime end = new DateTime(endTime);
|
List<List<String>> newList = Lists.partition(tagCodes, 10);
|
List<ListenableFuture<QueryResult>> taskList = new ArrayList<>();
|
ListeningExecutorService service =
|
MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(50));
|
|
final String sql = "select %s(value) as value, quality FROM FROM %s WHERE %s AND time >=%dms " +
|
"AND" +
|
" time <=%dms GROUP BY code fill(previous)";
|
for (List<String> tags : newList) {
|
ListenableFuture<QueryResult> futureTask = service.submit(() -> {
|
StringBuilder sb = new StringBuilder();
|
String codes = formatCode(tags);
|
sb.append(String.format(sql, queryType.name(),
|
TABLE_NAME, codes, begin.toDate().getTime(), end.toDate().getTime() + 999));
|
return builderResultValue(sb.toString());
|
});
|
taskList.add(futureTask);
|
}
|
List<RtdbResult> results = buildResult(taskList);
|
service.shutdown();
|
return results;
|
}
|
|
/**
|
* 插入实时数据
|
*
|
* @param tagValues 实时数据集合
|
* @return
|
*/
|
public Boolean storeData(List<RtdbResult> tagValues) {
|
Instant now = Instant.now();
|
for (RtdbResult tagValue : tagValues) {
|
tagValue.setTime(now);
|
}
|
return saveData(tagValues);
|
}
|
|
/**
|
* 插入历史数据
|
*
|
* @param tagValues 历史数据集合
|
* @return
|
*/
|
public Boolean insertData(List<RtdbResult> tagValues) {
|
return saveData(tagValues);
|
}
|
|
|
/**
|
* 删除测点
|
*
|
* @param tagCode 点位号
|
* @return
|
*/
|
public Boolean deleteTag(String tagCode) {
|
return false;
|
}
|
|
private Boolean saveData(List<RtdbResult> tagValues) {
|
BatchPoints.Builder builder = BatchPoints
|
.database(dataBaseName)
|
.retentionPolicy("autogen").precision(TimeUnit.SECONDS);
|
tagValues.forEach(tagValue -> {
|
Point point = Point
|
.measurement(TABLE_NAME)
|
.addField("value", tagValue.getValue())
|
.addField("quality", tagValue.getQuality())
|
.tag("code", tagValue.getTagCode())
|
.time(tagValue.getTime().getEpochSecond(), TimeUnit.SECONDS).build();
|
builder.point(point);
|
});
|
influxDB.write(builder.build());
|
return true;
|
}
|
|
private QueryResult builderResultValue(String sql) {
|
try {
|
// logger.error("计算数据的SQL是" + sql);
|
Query query = new Query(sql, dataBaseName);
|
return influxDB.query(query);
|
} catch (Throwable throwable) {
|
return new QueryResult();
|
}
|
}
|
|
private List<RtdbResult> buildResult(List<ListenableFuture<QueryResult>> taskList) {
|
List<RtdbResult> results = new ArrayList<>();
|
|
try {
|
for (QueryResult queryResult : Futures.successfulAsList(taskList).get()) {
|
try {
|
synchronized (this) {
|
if (queryResult != null) {
|
List<RtdbResult> rtdbResults = resultMapper.toPOJO(queryResult, RtdbResult.class);
|
results.addAll(rtdbResults);
|
}
|
}
|
} catch (Throwable throwable) {
|
}
|
}
|
} catch (InterruptedException | ExecutionException e) {
|
}
|
|
return results;
|
}
|
|
private String formatCode(List<String> codeList) {
|
return "(code='" + String.join("' or code='", codeList) + "')";
|
}
|
}
|
