InfluxDB使用教程：Java版InfluxDB工具类

前言：

上几讲，介绍了Influx在Linux和Windows上的使用之后，本节开始介绍Influx在Java中的使用，先提供一个InfluxDB Java API 封装的工具类，方便大家直接上手使用。

1.InfluxDB工具类

先奉上工具类，接下来介绍使用方法（更新于2021年11月08日）。

package com.common.utils.influxdb; 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 xuchao */ public class InfluxDBConnection { // 用户名 private String username; // 密码 private String password; // 连接地址 private String openurl; // 数据库名称 private String dbName; // 保留策略 private String retentionPolicy; private InfluxDB influxDB; public InfluxDBConnection(String username, String password, String url, String dbName, String retentionPolicy) { this.username = username; this.password = password; this.openurl = url; this.dbName = dbName; 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 days 保存天数 * @param replication 保存副本数量 * @param isDefault 是否设为默认保留策略 */ public void createRetentionPolicy(String dataBaseName, String policyName, int days, int replication, Boolean isDefault) { String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %sd REPLICATION %s ", policyName, dataBaseName, days, replication); if (isDefault) { sql = sql + " DEFAULT"; } query(sql); } / * 创建默认的保留策略 * * 策略名：hour，保存天数：30天，保存副本数量：1,设为默认保留策略 */ public void createDefaultRetentionPolicy() { String command = String .format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT", "hour", dbName, "30d", 1); this.query(command); } / * 查询 * * @param command 查询语句 * @return */ public QueryResult query(String command) { return influxDB.query(new Query(command, dbName)); } / * 插入 * * @param measurement 表 * @param tags 标签 * @param fields 字段 */ public void insert(String measurement, Map 
  
    tags, Map 
   
     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(dbName, retentionPolicy, builder.build()); } / * 批量写入测点 * * @param batchPoints */ public void batchInsert(BatchPoints batchPoints, TimeUnit timeUnit) { influxDB.write(batchPoints); // influxDB.enableGzip(); // influxDB.enableBatch(2000,100,timeUnit); // 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, TimeUnit timeUnit, final List 
    
      records) { influxDB.write(database, retentionPolicy, consistency, timeUnit, records); } / * 删除 * * @param command 删除语句 * @return 返回错误信息 */ public String deleteMeasurementData(String command) { QueryResult result = influxDB.query(new Query(command, dbName)); return result.getError(); } / * 关闭数据库 */ public void close() { influxDB.close(); } / * 构建Point * * @param measurement * @param time * @param fields * @return */ public Point pointBuilder(String measurement, long time, TimeUnit timeUnit, Map 
     
       tags, Map 
      
        fields) { Point point = Point.measurement(measurement).time(time, timeUnit).tag(tags).fields(fields).build(); return point; } } 
       
      
     
    
  

依赖的Jar包：

 
   
   
     org.influxdb 
    
   
     influxdb-java 
    
   
     2.10 
    
   

或者：

 
   
   
     org.influxdb 
    
   
     influxdb-java 
    
   
     2.21 
    
   

以上两个版本都经过测试可用。

二.使用工具类查询数据

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 
  
    > valueList = oneResult.getSeries().stream().map(Series::getValues) .collect(Collectors.toList()).get(0); if (valueList != null && valueList.size() > 0) { for (List 
    
  

取数据的时候，注意空值判断，本例将返回数据先进行判空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 
  
    tags = new HashMap 
   
     (); tags.put("tag1", "标签值"); Map 
    
      fields = new HashMap 
     
       (); fields.put("field1", "String类型"); // 数值型，InfluxDB的字段类型，由第一天插入的值得类型决定 fields.put("field2", 3.); // 时间使用毫秒为单位 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 
  
    tags = new HashMap 
   
     (); tags.put("tag1", "标签值"); Map 
    
      fields1 = new HashMap 
     
       (); fields1.put("field1", "abc"); // 数值型，InfluxDB的字段类型，由第一天插入的值得类型决定 fields1.put("field2", ); Map 
      
        fields2 = new HashMap 
       
         (); fields2.put("field1", "String类型"); fields2.put("field2", 3.); // 一条记录值 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,TimeUnit.MILLISECONDS); influxDBConnection.batchInsert(batchPoints2,TimeUnit.MILLISECONDS); } 
        
       
      
     
    
  

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

public static void main(String[] args) { InfluxDBConnection influxDBConnection = new InfluxDBConnection("admin", "admin", "1.1.1.1", "db-test", "hour"); Map 
  
    tags1 = new HashMap 
   
     (); tags1.put("tag1", "标签值"); Map 
    
      tags2 = new HashMap 
     
       (); tags2.put("tag2", "标签值"); Map 
      
        fields1 = new HashMap 
       
         (); fields1.put("field1", "abc"); // 数值型，InfluxDB的字段类型，由第一天插入的值得类型决定 fields1.put("field2", ); Map 
        
          fields2 = new HashMap 
         
           (); fields2.put("field1", "String类型"); fields2.put("field2", 3.); // 一条记录值。（注意：生产环境不要用System.currentTimeMillis()，因为数据量大会产生重复时间戳，导致数据丢失，要用数据自己的时间戳，这里只做演示） Point point1 = influxDBConnection.pointBuilder("表名", System.currentTimeMillis(), TimeUnit.MILLISECONDS,tags1, fields1); Point point2 = influxDBConnection.pointBuilder("表名", System.currentTimeMillis(), TimeUnit.MILLISECONDS,tags2, fields2); BatchPoints batchPoints1 = BatchPoints.database("db-test").tag("tag1", "标签值1") .retentionPolicy("hour").consistency(ConsistencyLevel.ALL).precision(TimeUnit.MILLISECONDS).build(); // 将两条记录添加到batchPoints中 batchPoints1.point(point1); BatchPoints batchPoints2 = BatchPoints.database("db-test").tag("tag2", "标签值2") .retentionPolicy("hour").consistency(ConsistencyLevel.ALL).precision(TimeUnit.MILLISECONDS).build(); // 将两条记录添加到batchPoints中 batchPoints2.point(point2); // 将不同的batchPoints序列化后，一次性写入数据库，提高写入速度 List 
          
            records = new ArrayList 
           
             (); records.add(batchPoints1.lineProtocol()); records.add(batchPoints2.lineProtocol()); // 将两条数据批量插入到数据库中 influxDBConnection.batchInsert("db-test", "hour", ConsistencyLevel.ALL,TimeUnit.MILLISECONDS, records); } 
            
           
          
         
        
       
      
     
    
  

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

总结：

Influx在Java中的读取和写入，还有批量写入数据的两种方式，今天都给大家介绍了，希望本工具类和案例可以给你带来帮助。