JAVA操作InfluxDB的一個Demo

一、基礎連接配接類

package com.test.repository.utils;

import com.test.domain.entry.bo.common.InfluxDbRow;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;

import java.util.List;
import java.util.concurrent.TimeUnit;


public class InfluxDBConnect {

    private String username;
    private String password;
    private String url;
    private String database;
    private int retentionDay;
    private int replicationCount;

    private InfluxDB influxDB;


    public InfluxDBConnect(String username, String password, String url, String database, int retentionDay, int replicationCount) {
        this.username = username;
        this.password = password;
        this.url = url;
        this.database = database;
        this.retentionDay = retentionDay;
        this.replicationCount = replicationCount;
    }

    /** 連接配接時序資料庫；獲得InfluxDB **/
    void connection() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(url, username, password);
        }
    }

    /**
     * 設定資料儲存政策
     * defalut 政策名 /database 資料庫名/ 30d 資料儲存時限30天/ 1  副本個數為1/ 結尾DEFAULT 表示 設為預設的政策
     */
    void createRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                "default", database, retentionDay + "d", replicationCount);
        this.query(command);
    }

    /**
     * 查詢
     * @param command 查詢語句
     * @return 查詢結果
     */
    QueryResult query(String command) {
        return influxDB.query(new Query(command, database));
    }

    /**
     * 插入
     */
    public void insert(InfluxDbRow influxDbRow) {
        if (influxDbRow == null) {
            return;
        }
        Point.Builder builder = Point.measurement(influxDbRow.getMeasurement());
        builder.tag(influxDbRow.getTags());
        builder.fields(influxDbRow.getFields());
        if (influxDbRow.getTimeSecond() != null) {
            builder.time(influxDbRow.getTimeSecond(), TimeUnit.SECONDS);
        }
        influxDB.write(database, "default", builder.build());
    }

    /**
     * 删除
     * @param command 删除語句
     * @return 傳回錯誤資訊
     */
    public String deleteMeasurementData(String command) {
        QueryResult result = influxDB.query(new Query(command, database));
        return result.getError();
    }

    /**
     * 建立資料庫
     * @param dbName 庫名稱
     */
    public void createDB(String dbName) {
        this.query("create database " + dbName);
    }

    /**
     * 删除資料庫
     * @param dbName
     */
    public void deleteDB(String dbName) {
        this.query("drop database " + dbName);
    }

    public void close() {
        this.influxDB.close();
    }

    /**
     * 指導導入
     * @param influxDbRows 行記錄
     */
    public void batchPointsImport(List<InfluxDbRow> influxDbRows) {
        if (influxDbRows == null || influxDbRows.size() == 0) {
            return;
        }
        BatchPoints batchPoints = BatchPoints.database(this.database).retentionPolicy("default").build();
        for (InfluxDbRow influxDbRow : influxDbRows) {
            if (influxDbRow.getTags().size() + influxDbRow.getFields().size() == 0) continue;
            Point.Builder builder = Point.measurement(influxDbRow.getMeasurement());
            builder.tag(influxDbRow.getTags());
            builder.fields(influxDbRow.getFields());
            if (influxDbRow.getTimeSecond() != null) {
                builder.time(influxDbRow.getTimeSecond(), TimeUnit.SECONDS);
            } else {
                builder.time(System.currentTimeMillis() / 1000, TimeUnit.SECONDS);
            }
            batchPoints.point(builder.build());
        }
        influxDB.write(batchPoints);
    }
}      

package com.test.repository.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;


@Configuration
@Slf4j
public class InfluxDBConnectConfig {

    @Value("${spring.influx.url}")
    private String url;
    @Value("${spring.influx.user}")
    private String username;
    @Value("${spring.influx.password}")
    private String password;
    @Value("${spring.influx.database}")
    private String database;
    @Value("${spring.influx.retentionDay}")
    private Integer retentionDay;
    @Value("${spring.influx.replicationCount}")
    private Integer replicationCount;

    @Bean
    @Scope("prototype")
    public InfluxDBConnect influxDBConnectFactory() {
        if (this.retentionDay == null) this.retentionDay = 30;
        if (this.replicationCount == null) this.replicationCount = 1;
        return new InfluxDBConnect(username, password, url, database, retentionDay, replicationCount);
    }

}      

宋興柱(Sindrol)：轉載内容，請标明出處，謝謝！源文來自 寶貝雲知識分享：https://www.dearcloud.cn