57°

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数据库的主要作用是监控。

java使用influxDB时序数据库

本文由【S】发布于开源中国,原文链接:https://my.oschina.net/heker/blog/3160271

全部评论: 0

    我有话说: