'java使用influxDB時序數據庫'

數據庫 Java pong 半條命A 2019-08-12

本人寫的這篇文章主要是介紹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數據庫的主要作用是監控。