InfluxDB在Java中的使用

InfluxDB在Java中的使用

InfluxDB（时序数据库），常用的一种使用场景：监控数据统计。每毫秒记录一下电脑内存的使用情况，然后就可以根据统计的数据，利用图形化界面（InfluxDB V1一般配合Grafana）制作内存使用情况的折线图；

可以理解为按时间记录一些数据（常用的监控数据、埋点统计数据等），然后制作图表做统计；

依赖的Jar包：

<dependency> 	
		<groupId>org.influxdb</groupId> 	
		<artifactId>influxdb-java</artifactId> 	
		<version>2.10</version> 
</dependency>

InfluxDB工具类

package com.common.utils.influxdb;

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;

/**

 * 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();
      } 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;
      }

}

二.使用工具类查询数据

InfluxDB支持一次查询多个SQL，SQL之间用分号;隔开即可。下面只演示下，只有一条SQL的情况下，怎么解析查询返回的结果集。

public static void main(String[] args) {
		InfluxDBConnection influxDBConnection = new InfluxDBConnection("admin", "admin", "1.1.1.1", "db-test", "hour");
		QueryResult results = influxDBConnection
				.query("SELECT * FROM measurement where name = '大脑补丁'  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 用取出的字段做你自己的业务逻辑……
				}
			}
		}
	}

取数据的时候，注意空值判断，本例将返回数据先进行判空oneResult.getSeries() != null，然后调用oneResult.getSeries().getValues().get(0)获取到第一条SQL的返回结果集，然后遍历valueList，取出每条记录中的目标字段值。

InfluxDB封装的结果集有点深，主要是由于支持多条SQL一次性查询，可以提高查询速度，这个地方有别于关系型数据库的使用。

二.使用InfluxDB工具类，插入单条数据

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

	public static void main(String[] args) {
		InfluxDBConnection influxDBConnection = new InfluxDBConnection("admin", "admin", "1.1.1.1", "db-test", "hour");
		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工具类，批量写入数据的两种方式

注：使用这两种种方式，要就这两条数据都写入到同一数据库下且tag相同,若tag不相同，需将它们放到不同的BatchPoint对象中，否则会出现数据写入错乱问题。

方式一：通过BatchPoints组装数据后，循环插入数据库。

public static void main(String[] args) {
		InfluxDBConnection influxDBConnection = new InfluxDBConnection("admin", "admin", "1.1.1.1", "db-test", "hour");
		Map<String, String> tags = new HashMap<String, String>();
		tags.put("tag1", "标签值");
		Map<String, Object> fields1 = new HashMap<String, Object>();
		fields1.put("field1", "abc");
		// 数值型，InfluxDB的字段类型，由第一天插入的值得类型决定
		fields1.put("field2", 123456);
		Map<String, Object> fields2 = new HashMap<String, Object>();
		fields2.put("field1", "String类型");
		fields2.put("field2", 3.141592657);
		// 一条记录值
		Point point1 = influxDBConnection.pointBuilder("表名", System.currentTimeMillis(), tags, fields1);
		Point point2 = influxDBConnection.pointBuilder("表名", System.currentTimeMillis(), tags, fields2);
		// 将两条记录添加到batchPoints中
		BatchPoints batchPoints1 = BatchPoints.database("db-test").tag("tag1", "标签值1").retentionPolicy("hour")
				.consistency(ConsistencyLevel.ALL).build();
		BatchPoints batchPoints2 = BatchPoints.database("db-test").tag("tag2", "标签值2").retentionPolicy("hour")
				.consistency(ConsistencyLevel.ALL).build();
		batchPoints1.point(point1);
		batchPoints2.point(point2);
		// 将两条数据批量插入到数据库中
		influxDBConnection.batchInsert(batchPoints1);
		influxDBConnection.batchInsert(batchPoints2);
	}

方式二：通过BatchPoints组装数据，序列化后，一次性插入数据库。

public static void main(String[] args) {
		InfluxDBConnection influxDBConnection = new InfluxDBConnection("admin", "admin", "1.1.1.1", "db-test", "hour");
		Map<String, String> tags1 = new HashMap<String, String>();
		tags1.put("tag1", "标签值");
		Map<String, String> tags2 = new HashMap<String, String>();
		tags2.put("tag2", "标签值");
		Map<String, Object> fields1 = new HashMap<String, Object>();
		fields1.put("field1", "abc");
		// 数值型，InfluxDB的字段类型，由第一天插入的值得类型决定
		fields1.put("field2", 123456);
		Map<String, Object> fields2 = new HashMap<String, Object>();
		fields2.put("field1", "String类型");
		fields2.put("field2", 3.141592657);
		// 一条记录值
		Point point1 = influxDBConnection.pointBuilder("表名", System.currentTimeMillis(), tags1, fields1);
		Point point2 = influxDBConnection.pointBuilder("表名", System.currentTimeMillis(), tags2, fields2);
		BatchPoints batchPoints1 = BatchPoints.database("db-test").tag("tag1", "标签值1")
				.retentionPolicy("hour").consistency(ConsistencyLevel.ALL).build();
		// 将两条记录添加到batchPoints中
		batchPoints1.point(point1);
		BatchPoints batchPoints2 = BatchPoints.database("db-test").tag("tag2", "标签值2")
				.retentionPolicy("hour").consistency(ConsistencyLevel.ALL).build();
		// 将两条记录添加到batchPoints中
		batchPoints2.point(point2);
		// 将不同的batchPoints序列化后，一次性写入数据库，提高写入速度
		List<String> records = new ArrayList<String>();
		records.add(batchPoints1.lineProtocol());
		records.add(batchPoints2.lineProtocol());
		// 将两条数据批量插入到数据库中
		influxDBConnection.batchInsert("db-test", "hour", ConsistencyLevel.ALL, records);
	}

推荐使用第二种方式，属于一个数据库的数据，可以一次性批量写入，写入速度最快。

版权声明：本文为LHZ_CH原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。