InfluxDB【InfluxDB数据库操作】

InfluxDB数据库连接操作类

/** Copyright (c) 2010-2020. */package com.sjasoft.cloud.admin.inflixdbconn;import org.influxdb.InfluxDB;
import org.influxdb.InfluxDB.ConsistencyLevel;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.*;
import org.influxdb.dto.Point.Builder;import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;/*** InfluxDB数据库连接操作类** @author 大脑补丁*/
public class InfluxDBConnection {// 用户名private String username;// 密码private String password;// 连接地址private String openurl;// 数据库private String database;// 保留策略private String retentionPolicy;private InfluxDB influxDB;public InfluxDBConnection(String username, String password, String openurl, String database,String retentionPolicy) {this.username = username;this.password = password;this.openurl = openurl;this.database = database;this.retentionPolicy = retentionPolicy == null || retentionPolicy.equals("") ? "autogen" : retentionPolicy;influxDbBuild();}/*** 创建数据库** @param dbName*/@SuppressWarnings("deprecation")public void createDB(String dbName) {influxDB.createDatabase(dbName);}/*** 删除数据库** @param dbName*/@SuppressWarnings("deprecation")public void deleteDB(String dbName) {influxDB.deleteDatabase(dbName);}/*** 测试连接是否正常** @return true 正常*/public boolean ping() {boolean isConnected = false;Pong pong;try {pong = influxDB.ping();if (pong != null) {isConnected = true;}} catch (Exception e) {e.printStackTrace();}return isConnected;}/*** 连接时序数据库 ，若不存在则创建** @return*/public InfluxDB influxDbBuild() {if (influxDB == null) {influxDB = InfluxDBFactory.connect(openurl, username, password);}try {// if (!influxDB.databaseExists(database)) {// influxDB.createDatabase(database);// }} catch (Exception e) {// 该数据库可能设置动态代理，不支持创建数据库// e.printStackTrace();}influxDB.setLogLevel(InfluxDB.LogLevel.NONE);return influxDB;}/*** 创建自定义保留策略** @param policyName*            策略名* @param duration*            保存天数* @param replication*            保存副本数量* @param isDefault*            是否设为默认保留策略*/public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault) {String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,database, duration, replication);if (isDefault) {sql = sql + " DEFAULT";}this.query(sql);}/*** 创建默认的保留策略** @param ：default，保存天数：30天，保存副本数量：1*            设为默认保留策略*/public void createDefaultRetentionPolicy() {String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT","default", database, "30d", 1);this.query(command);}/*** 查询** @param command*            查询语句* @return*/public QueryResult query(String command) {return influxDB.query(new Query(command, database));}/*** 插入** @param measurement*            表* @param tags*            标签* @param fields*            字段*/public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields, long time,TimeUnit timeUnit) {Builder builder = Point.measurement(measurement);builder.tag(tags);builder.fields(fields);if (0 != time) {builder.time(time, timeUnit);}influxDB.write(database, retentionPolicy, builder.build());}/*** 批量写入测点** @param batchPoints*/public void batchInsert(BatchPoints batchPoints) {influxDB.write(batchPoints);// influxDB.enableGzip();// influxDB.enableBatch(2000,100,TimeUnit.MILLISECONDS);// influxDB.disableGzip();// influxDB.disableBatch();}/*** 批量写入数据** @param database*            数据库* @param retentionPolicy*            保存策略* @param consistency*            一致性* @param records*            要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）*/public void batchInsert(final String database, final String retentionPolicy, final ConsistencyLevel consistency,final List<String> records) {influxDB.write(database, retentionPolicy, consistency, records);}/*** 删除** @param command*            删除语句* @return 返回错误信息*/public String deleteMeasurementData(String command) {QueryResult result = influxDB.query(new Query(command, database));return result.getError();}/*** 关闭数据库*/public void close() {influxDB.close();}/*** 构建Point** @param measurement* @param time* @param fields* @return*/public Point pointBuilder(String measurement, long time, Map<String, String> tags, Map<String, Object> fields) {Point point = Point.measurement(measurement).time(time, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();return point;}}

InfluxDBUtil【influxDb工具类（将数据库中的数据提取至influx监测类集合中）】

package com.sjasoft.cloud.admin.inflixdbconn;import com.sjasoft.cloud.admin.dto.emsapp.EnvMonitorDataVo;
import org.springframework.util.CollectionUtils;
import java.util.ArrayList;
import java.util.List;public class InfluxDBUtil {public static List<EnvMonitorDataVo> getEnvMonitorData(List<List<Object>> valueList){List<EnvMonitorDataVo> monitorDataList = new ArrayList<>();EnvMonitorDataVo envMonitorDataVo;
//            List<List<Object>> valueList = resultSeries.stream()
//                    .map(QueryResult.Series::getValues).collect(Collectors.toList()).get(0);if(!CollectionUtils.isEmpty(valueList)){for (List<Object> value : valueList) {envMonitorDataVo =new EnvMonitorDataVo();// 数据库中字段1取值//String field1 = value.get(0) == null ? null : value.get(0).toString();// 数据库中字段2取值//String field2 = value.get(1) == null ? null : value.get(1).toString();// TODO 将数据库中的数据提取至influx监测类中//时间envMonitorDataVo.setTime(value.get(0) == null ? null : value.get(0).toString());//坐标类型envMonitorDataVo.setCoordinateType(value.get(1) == null ? 0 : Float.parseFloat(value.get(1).toString()));//湿度envMonitorDataVo.setHum(value.get(2) == null ? 0.0f : Float.parseFloat(value.get(2).toString()));//纬度envMonitorDataVo.setLat(value.get(3) == null ? 0.0f : Float.parseFloat(value.get(3).toString()));//经度envMonitorDataVo.setLng(value.get(4) == null ? 0.0f : Float.parseFloat(value.get(4).toString()));//噪音envMonitorDataVo.setNoise(value.get(5) == null ? 0.0f : Float.parseFloat(value.get(5).toString()));//Pm10envMonitorDataVo.setPm10(value.get(6) == null ? 0.0f : Float.parseFloat(value.get(6).toString()));//Pm2.5envMonitorDataVo.setPm25(value.get(7) == null ? 0.0f : Float.parseFloat(value.get(7).toString()));//气压envMonitorDataVo.setPressure(value.get(8) == null ? 0.0f : Float.parseFloat(value.get(8).toString()));//继电器状态envMonitorDataVo.setRelayStatus(value.get(9) == null ? null : value.get(9).toString());//TSPenvMonitorDataVo.setTsp(value.get(10) == null ? 0.0f : Float.parseFloat(value.get(10).toString()));//温度envMonitorDataVo.setTem(value.get(11) == null ? 0.0f : Float.parseFloat(value.get(11).toString()));//风向系数 0北风 1东北风 2东风 3 东南风 4南风 5西南风 6西风 7西北风envMonitorDataVo.setWindDirectionCoefficient(value.get(12) == null ? 0.0f : Float.parseFloat(value.get(12).toString()));//风向度数envMonitorDataVo.setWindDirectionDegree(value.get(13) == null ? 0.0f : Float.parseFloat(value.get(13).toString()));//风力envMonitorDataVo.setWindPower(value.get(14) == null ? 0.0f : Float.parseFloat(value.get(14).toString()));//风速envMonitorDataVo.setWindSpeed(value.get(15) == null ? 0.0f : Float.parseFloat(value.get(15).toString()));//设备idenvMonitorDataVo.setDeviceId(value.get(16) == null ? null : value.get(16).toString());monitorDataList.add(envMonitorDataVo);}}return monitorDataList;}}

监测信息使用类

package com.sjasoft.cloud.admin.service.emsapp;import com.alibaba.fastjson.JSON;
import com.sjasoft.cloud.admin.dto.emsapp.EnvMonitorDataVo;
import com.sjasoft.cloud.admin.dto.emsapp.EnvMonitorParam;
import com.sjasoft.cloud.admin.inflixdbconn.InfluxDBConnection;
import com.sjasoft.cloud.admin.inflixdbconn.InfluxDBUtil;
import org.influxdb.dto.QueryResult;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;import java.lang.management.MonitorInfo;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;/**
* @Author lufei
* @Description 环境监测
* @Date 2021/2/4 下午 05:19
**/
@Service
public class EnvMonitorService {private static Logger logger = LoggerFactory.getLogger(EnvMonitorService.class);@Value("${InfluxDBConnection.url}")private  String url;@Value("${InfluxDBConnection.username}")private  String username;@Value("${InfluxDBConnection.password}")private  String password;@Value("${time.zone}")private String timeZone;@Value("${InfluxDBConnection.dataBaseName}")private String influxDbDataBaseNm;/*** 查看环境监测设备是否在线及在线时部门监测信息* @param envMonitorParam* @return*/public EnvMonitorDataVo getEnvMonitorIsOnline(@NotNull EnvMonitorParam envMonitorParam){InfluxDBConnection influxDBConnection = new InfluxDBConnection(username, password, url, influxDbDataBaseNm, "default");//'"+deviceId+"'EnvMonitorDataVo monitorDataVo = new EnvMonitorDataVo();try {//String sql = "SELECT last(DATA_VALUE) from \"t_device_data\"  WHERE device_id='"+envMonitorParam.getDeviceId()+"' AND  time > now() - 2m GROUP BY DATA_TYPE tz( ' "+timeZone+" ')";String sqlIsOnline = "SELECT CoordinateType,Hum,Lat,Lng,Noise,Pm10,Pm25,Pressure,RelayStatus,Tsp,Tem,WindDirectionCoefficient,WindDirectionDegree,WindPower,WindSpeed,device_id from \"Renke-RS-ZSYC\"  WHERE device_id='"+envMonitorParam.getDeviceId()+"' AND  time > now() - 8h order by time desc limit 1";logger.info("=====查询是否在线执行的sql:{}==========",sqlIsOnline);/************************判断是否在线**************************/QueryResult queryIsOnline = influxDBConnection.query(sqlIsOnline);logger.info("=====influxDB====查询是否在线===成功====！！！！=====");QueryResult.Result isOnlineResult = queryIsOnline.getResults().get(0);if (CollectionUtils.isEmpty(isOnlineResult.getSeries())) {logger.info("=====influxDB====【不在线】=======！！！！=====");}else {List<List<Object>> valueList = isOnlineResult.getSeries().stream().map(QueryResult.Series::getValues).collect(Collectors.toList()).get(0);List<EnvMonitorDataVo> monitorDataList = InfluxDBUtil.getEnvMonitorData(valueList);if(!CollectionUtils.isEmpty(monitorDataList)){BeanUtils.copyProperties(monitorDataList.get(0),monitorDataVo);logger.info("==getEnvMonitorIsOnline==返回：={}",JSON.toJSONString(monitorDataVo));monitorDataVo.setIsOnline("1");logger.info("=====influxDB====在线=======！！！！=====");}}}catch (Exception e){logger.error("======================InfluxDB===调用失败！=======================");e.printStackTrace();}finally {influxDBConnection.close();}return monitorDataVo;}/*** 环境监测实时数据查看* @param envMonitorParam* @return*/public EnvMonitorDataVo getEnvMonitorRealTimeData(@NotNull EnvMonitorParam envMonitorParam){InfluxDBConnection influxDBConnection = new InfluxDBConnection(username, password, url, influxDbDataBaseNm, "default");//'"+deviceId+"'EnvMonitorDataVo monitorDataVo = new EnvMonitorDataVo();try {//String sql = "SELECT last(DATA_VALUE) from \"t_device_data\"  WHERE device_id='"+envMonitorParam.getDeviceId()+"' AND  time > now() - 2m GROUP BY DATA_TYPE tz( ' "+timeZone+" ')";String sql = "SELECT CoordinateType,Hum,Lat,Lng,Noise,Pm10,Pm25,Pressure,RelayStatus,Tsp,Tem,WindDirectionCoefficient,WindDirectionDegree,WindPower,WindSpeed,device_id from \"Renke-RS-ZSYC\" WHERE device_id='"+envMonitorParam.getDeviceId()+"' order by time desc limit 1";logger.info("=====查询监测信息执行的sql:{}==========",sql);QueryResult queryResult = influxDBConnection.query(sql);logger.info("=====influxDB====查询监测信息===成功====！！！！=====");QueryResult.Result oneResult = queryResult.getResults().get(0);if (!CollectionUtils.isEmpty(oneResult.getSeries())) {List<List<Object>> valueList = oneResult.getSeries().stream().map(QueryResult.Series::getValues).collect(Collectors.toList()).get(0);List<EnvMonitorDataVo> monitorDataList = InfluxDBUtil.getEnvMonitorData(valueList);if(!CollectionUtils.isEmpty(monitorDataList)){BeanUtils.copyProperties(monitorDataList.get(0),monitorDataVo);logger.info("==getEnvMonitorIsOnline==返回：={}",JSON.toJSONString(monitorDataVo));}}}catch (Exception e){logger.error("======================InfluxDB===调用失败！=======================");e.printStackTrace();}finally {influxDBConnection.close();}return monitorDataVo;}/*** 根据设备编号查看环境监测一天内的某项采样所有数值* 此次的一天不可直接推日期【使用24小时计算】* @param envMonitorParam* @return*/public List<EnvMonitorDataVo> getEnvMonitorOneDaySampInfo(@NotNull EnvMonitorParam envMonitorParam){InfluxDBConnection influxDBConnection = new InfluxDBConnection(username, password, url, influxDbDataBaseNm, "default");//'"+deviceId+"'List<EnvMonitorDataVo> monitorDataList = new ArrayList<>();EnvMonitorDataVo monitorDataVo = null;try {//String sql = "SELECT last(DATA_VALUE) from \"t_device_data\"  WHERE device_id='"+envMonitorParam.getDeviceId()+"' AND  time > now() - 2m GROUP BY DATA_TYPE tz( ' "+timeZone+" ')";String sql = "SELECT CoordinateType,Hum,Lat,Lng,Noise,Pm10,Pm25,Pressure,RelayStatus,Tsp,Tem,WindDirectionCoefficient,WindDirectionDegree,WindPower,WindSpeed,device_id from \"Renke-RS-ZSYC\" WHERE time > now() - 40h  AND  device_id='"+envMonitorParam.getDeviceId()+"' order by time desc";logger.info("=====查询监测信息执行的sql:{}==========",sql);QueryResult queryResult = influxDBConnection.query(sql);logger.info("=====influxDB====查询监测信息===成功====！！！！=====");//QueryResult.Result oneResult = queryResult.getResults().get(0);//获取一天内数据集合QueryResult.Result resultList = queryResult.getResults().get(0);if(!CollectionUtils.isEmpty(resultList.getSeries())){List<List<Object>> valueList = resultList.getSeries().stream().map(QueryResult.Series::getValues).collect(Collectors.toList()).get(0);monitorDataList = InfluxDBUtil.getEnvMonitorData(valueList);}if(CollectionUtils.isEmpty(monitorDataList)){logger.info("=====InfluxDB===：=======未查到数据=========");}else {logger.info("=====InfluxDB===：=======查到{}条数据=========",monitorDataList.size());}}catch (Exception e){logger.error("======================InfluxDB===调用失败！=======================");e.printStackTrace();}finally {influxDBConnection.close();}return monitorDataList;}}

application.properties简单配置


# 亚洲时区
time.zone=Asia/Shanghai
# 服务器的时序数据库
InfluxDBConnection.url=http://192.168.2.50:8086
InfluxDBConnection.username=csems
InfluxDBConnection.password=qaz123/*-+
# 数据库名
InfluxDBConnection.dataBaseName = eam-buildingsite# 组织表中图片前缀
orgImage.imagePrefix = E:\\360Downloads\testImage

InfluxDB【InfluxDB数据库操作】相关推荐

ASP.NET Core2读写InfluxDB时序数据库
在我们很多应用中会遇到有一种基于一系列时间的数据需要处理,通过时间的顺序可以将这些数据点连成线,再通过数据统计后可以做成多纬度的报表,也可通过机器学习来实现数据的预测告警.而时序数据库就是用于存放管理 ...
influxDB时序数据库的使用
简介时序数据库时序数据库全称为时间序列数据库.时间序列数据库主要用于指处理带时间标签(按照时间的顺序变化,即时间序列化)的数据,带时间标签的数据也称为时间序列数据. 时间序列数据主要由电力行业.化 ...
emq数据储存到mysql_EMQ X 规则引擎系列（三）存储消息到 InfluxDB 时序数据库
前言 InfluxDB 是一个用于存储和分析时间序列数据的开源数据库,内置 HTTP API,类 SQL 语句的支持和无结构的特性对使用者而言都非常友好.它强大的数据吞吐能力以及稳定的性能表现使其非常 ...
基于docker的influxdb时序数据库搭建
docker项目部署 1. docker influxdb时序数据库搭建方法一.docker-compose运行方法二.直接启动容器服务 1. docker influxdb时序数据库搭建在Ub ...
Python的数据库操作（Sqlalchemy）
ORM 全称 Object Relational Mapping, 翻译过来叫对象关系映射.简单的说,ORM 将数据库中的表与面向对象语言中的类建立了一种对应关系.这样,我们要操作数据库,数据库中的表 ...
便捷，轻巧的Groovy数据库操作
本文主要介绍Groovy对数据的CRUD操作,熟悉groovy.sql包,测试使用的数据库是H2. 1.数据库连接配置 //数据库连接配置 def db = [url:'jdbc:h2:mem:gro ...
Django模型之数据库操作-查询
六.数据库操作-查询 6.1基础条件查询 1 基本查询 get查询单一结果,如果不存在会抛出模型类.DoesNotExist异常. all查询多个结果.[返回的所有结果是列表] count查询结果数量 ...
命令行客户端MySQL基本命令的使用（登录、登出、数据库操作的SQL语句、表结构的SQL语句、表数据操作的SQL语句）
1. 登录和登出数据库登录数据库: 输入下面命令: mysql -uroot -p 说明: -u 后面是登录的用户名 [写成-u root也是可以的] -p 后面是登录密码, 如果不填写, 回车之 ...
.ne中的控制器循环出来的数据如何显示在视图上_Web程序设计-ASP.NET MVC4数据库操作实例...
ASP.NET MVC4数据库操作实例之前文章介绍了MVC4与Pure框架结合进行的网页设计过程中如何定义控制器.方法.模型.视图等.并使用实例进行了简单说明.本文将在此基础上进一步说明如何使用MV ...

InfluxDB【InfluxDB数据库操作】

InfluxDB数据库连接操作类

InfluxDBUtil【influxDb工具类（将数据库中的数据提取至influx监测类集合中）】

监测信息使用类

application.properties简单配置

InfluxDB【InfluxDB数据库操作】相关推荐

最新文章

热门文章