java使用influxDB时序数据库

本人写的这篇文章主要是介绍java如何使用influxDB时序数据库,具有一定的参考价值,感兴趣的小伙伴们可以参考一下。

废话不多说,直接上代码:

1、pom.xml引入相关jar文件,如下:

<!-- 引入influxdb依赖 --> <dependency> <groupId>org.influxdb</groupId> <artifactId>influxdb-java</artifactId> <version>2.8</version> </dependency>

2、influxDB工具类封装:

package com.mt.core.util;import java.util.List;import java.util.Map;import java.util.concurrent.TimeUnit;import org.influxdb.InfluxDB;import org.influxdb.InfluxDB.ConsistencyLevel;import org.influxdb.InfluxDBFactory;import org.influxdb.dto.BatchPoints;import org.influxdb.dto.Point;import org.influxdb.dto.Point.Builder;import org.influxdb.dto.Pong;import org.influxdb.dto.Query;import org.influxdb.dto.QueryResult;import lombok.Data;

/** * InfluxDB数据库连接操作类 * * @author Simon */

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(); } finally { influxDB.setRetentionPolicy(retentionPolicy); } 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; }

}

3、查询数据

InfluxDB支持一次查询多个SQL,SQL之间用逗号隔开即可。

public static void main(String[] args) { InfluxDBConnection influxDBConnection = new InfluxDBConnection("root", "Password01", "localhost", "devops", "tk_test"); QueryResult results = influxDBConnection .query("SELECT * FROM mt order by time desc limit 1000"); //results.getResults()是同时查询多条SQL语句的返回值,此处我们只有一条SQL,所以只取第一个结果集即可。 Result oneResult = results.getResults().get(0); if (oneResult.getSeries() != null) { List<List<Object>> valueList = oneResult.getSeries().stream().map(Series::getValues) .collect(Collectors.toList()).get(0); if (valueList != null && valueList.size() > 0) { for (List<Object> value : valueList) { Map<String, String> map = new HashMap<String, String>(); // 数据库中字段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 用取出的字段做你自己的业务逻辑…… } } } }

4、插入数据

InfluxDB的字段类型,由第一条插入的值得类型决定;tags的类型只能是String型,可以作为索引,提高检索速度。

public static void main(String[] args) { InfluxDBConnection influxDBConnection = new InfluxDBConnection("root", "Password01", "localhost", "devops", "tk_test"); Map<String, String> tags = new HashMap<String, String>(); tags.put("tag1", "标签值"); Map<String, Object> fields = new HashMap<String, Object>(); fields.put("field1", "String类型"); // 数值型,InfluxDB的字段类型,由第一天插入的值得类型决定 fields.put("field2", 3.141592657); // 时间使用毫秒为单位 influxDBConnection.insert("表名", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS); }

总结:influxdb具有很强地并发写入能力,我没有做过相关的测试,但根据与其他使用者的沟通交流得知,一般主流配置下,每秒数万次的写入请求是非常轻松的。因为influxdb的机制,如此并发写入能力需要足够容量与速度的内存支持。更重要的一点,可以理解在influxdb中维护了许多时间轴,而数据库名、存储策略、measurement(类似mysql的表)名与tag名一起作为时间轴的标记(series)。也就是说,假设你把一个用户的数据复制并存储了两份,存在相同的数据库中,存在相同的表中,只不过第一份数据的保存策略是29天,第二份数据的保存策略是30天。那么也会被当作两份series来维护。而series的数目,是有上限的。

influxdb数据库的主要作用是监控。

原文地址:https://www.cnblogs.com/mengtaoadmin/p/11184043.html

时间: 2024-10-11 16:35:31

java使用influxDB时序数据库的相关文章

互联网级监控系统必备-时序数据库之Influxdb技术

时间序列数据库,简称时序数据库,Time Series Database,一个全新的领域,最大的特点就是每个条数据都带有Time列. 时序数据库到底能用到什么业务场景,答案是:监控系统. Baidu一下,互联网监控系统,大家会发现小米.饿了吗等互联网巨头都在用时序数据库实现企业级的互联网监控系统. 很多人会说,用Zabbix不就搞定了,其实不是这样的,简单的主机资源监控.网络监控.小规模的部署环境,Zabbix能搞定. 如果在IDC 上千台服务器环境下,分布式应用架构.各种中间件,这种情况下我们

互联网级监控系统必备-时序数据库之Influxdb集群及踩过的坑

上篇博文中,我们介绍了做互联网级监控系统的必备-Influxdb的关键特性.数据读写.应用场景: 互联网级监控系统必备-时序数据库之Influxdb 本文中,我们介绍Influxdb数据库集群的搭建,同时分享一下我们使用集群遇到的坑! 一.环境准备 同一网段内,3个CentOS 节点,相互可以ping通 3个节点CentOS配置Hosts文件,相互可以解析主机名 Azure 虚拟机启用root用户 influxdb-0.10.3-1.x86_64.rpm 设置端口8083 8086 8088 8

时序数据库InfluxDB使用详解

InfluxDB是一个开源的时序数据库,使用GO语言开发,特别适合用于处理和分析资源监控数据这种时序相关数据.而InfluxDB自带的各种特殊函数如求标准差,随机取样数据,统计数据变化比等,使数据统计和实时分析变得十分方便.在我们的容器资源监控系统中,就采用了InfluxDB存储cadvisor的监控数据.本文对InfluxDB的基本概念和一些特色功能做一个详细介绍,内容主要是翻译整理自官网文档,如有错漏,请指正. 来源:https://www.jianshu.com/p/a1344ca86e9

时序数据库InfluxDB

一.什么是InfluxDB? InfluxDB是一款用Go语言编写的开源分布式时序.事件和指标数据库,无需外部依赖.该数据库现在主要用于存储涉及大量的时间戳数据,如DevOps监控数据,APP metrics, loT传感器数据和实时分析数据. InfluxDB特征: – 无结构(无模式):可以是任意数量的列(tags). – 可以设置metric的保存时间. – 支持与时间有关的相关函数(如min.max.sum.count.mean.median等),方便统计. – 支持存储策略:可以用于数

时序数据库连载系列: 时序数据库一哥InfluxDB之存储机制解析

InfluxDB 的存储机制解析 本文介绍了InfluxDB对于时序数据的存储/索引的设计.由于InfluxDB的集群版已在0.12版就不再开源,因此如无特殊说明,本文的介绍对象都是指 InfluxDB 单机版 1. InfluxDB 的存储引擎演进 尽管InfluxDB自发布以来历时三年多,其存储引擎的技术架构已经做过几次重大的改动, 以下将简要介绍一下InfluxDB的存储引擎演进的过程. 1.1 演进简史 版本0.9.0之前 **基于 LevelDB的LSMTree方案** 版本0.9.0

时序数据库InfluxDB(I)- 搭建与采集信息demo操作

搭建环境:vmware workstation pro15.5.0, ubuntu18.04.3 实践时间:2019.10.12-10.27 (一)时序数据库InfluxDB准备 (1)安装 曾出现问题: 解决方法为: (2)配置文件的位置: 可进行InfluxDB缓存大小等参数的设置. (3)服务器端启动: 以后台运行的方式启动了InfluxDB的服务端. (4)客户端启动:登入InfluxDB服务器 将路径加入环境变量中,方便任意地方使用InfluxDB: Or: 特别提醒data和WAL的

[Go] 时序数据库influxdb的安装

日志类的数据时候存储在时序数据库中,下面就是时序数据库influxdb的安装 curl -sL https://repos.influxdata.com/influxdb.key | apt-key add -source /etc/lsb-releaseecho "deb https://repos.influxdata.com/${DISTRIB_ID,,} ${DISTRIB_CODENAME} stable" | tee /etc/apt/sources.list.d/infl

Java版InfluxDB工具类

InfluxDB工具类 1 package com.influxdb.test; 2 3 import java.util.Map; 4 5 import org.influxdb.InfluxDB; 6 import org.influxdb.InfluxDBFactory; 7 import org.influxdb.dto.Point; 8 import org.influxdb.dto.Point.Builder; 9 import org.influxdb.dto.Query; 10

[转帖]几个时序数据库

https://www.cnblogs.com/harrychinese/p/time_series_db.html ================================可用作时序的数据库:================================[时序]TimescaleDB, 基于 PostgreSQL, 支持 SQL.[时序]KairosDB, 基于 Cassandra, 不支持 SQL. [通用]CrateDB, 基于 Elastic Search, 但支持ANSI S