Spring boot使用influxDB總結

項目中須要存放大量設備日誌,且須要對其進行簡單的數據分析,信息提取工做.
結合衆多考量因素,項目決定使用時序數據庫中的領頭羊InfluxDB.

引入依賴

項目中使用influxdb-java,在pom文件中添加以下依賴(github地址:https://github.com/influxdata/influxdb-java):

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

application.yaml文件配置以下所示(請按照實際狀況填寫):

spring:
  influx:
    url: *
    password: admin
    user: 123
    database: log_management

配置

(1) 建立配置類

@Configuration
public class InfluxDbConfig {

    @Value("${spring.influx.url:''}")
    private String influxDBUrl;

    @Value("${spring.influx.user:''}")
    private String userName;

    @Value("${spring.influx.password:''}")
    private String password;

    @Value("${spring.influx.database:''}")
    private String database;

    @Bean
    public InfluxDbUtils influxDbUtils() {
        return new InfluxDbUtils(userName, password, influxDBUrl, database, "");
    }
}

@Data
 public class InfluxDbUtils {
    private String userName;
    private String password;
    private String url;
    public String database;
    private String retentionPolicy;
    // InfluxDB實例
    private InfluxDB influxDB;

    // 數據保存策略
    public static String policyNamePix = "logRetentionPolicy_";

    public InfluxDbUtils(String userName, String password, String url, String database,
                         String retentionPolicy) {
        this.userName = userName;
        this.password = password;
        this.url = url;
        this.database = database;
        this.retentionPolicy = retentionPolicy == null || "".equals(retentionPolicy) ? "autogen" : retentionPolicy;
        this.influxDB = influxDbBuild();
    }

    /**
     * 鏈接數據庫 ，若不存在則建立
     *
     * @return influxDb實例
     */
    private InfluxDB influxDbBuild() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(url, userName, password);
        }
        try {
            createDB(database);
            influxDB.setDatabase(database);
        } catch (Exception e) {
            log.error("create influx db failed, error: {}", e.getMessage());
        } finally {
            influxDB.setRetentionPolicy(retentionPolicy);
        }
        influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
        return influxDB;
    }
 }

構建實體類

InfluxDB中,measurement對應於傳統關係型數據庫中的table(database爲配置文件中的log_management).
InfluxDB裏存儲的數據稱爲時間序列數據,時序數據有零個或多個數據點.
數據點包括time(一個時間戳)，measurement(例如logInfo)，零個或多個tag，其對應於level,module,device_id),至少一個field(即日誌內容,msg=something error).
InfluxDB會根據tag數值創建時間序列(所以tag數值不能選取諸如UUID做爲特徵值,易致使時間序列過多,致使InfluxDB崩潰),並創建相應索引,以便優化諸如查詢速度.

@Builder
@Data
@Measurement(name = "logInfo")
public class LogInfo {

    // Column中的name爲measurement中的列名
    // 此外,須要注意InfluxDB中時間戳均是以UTC時保存,在保存以及提取過程當中須要注意時區轉換
    @Column(name = "time")
    private String time;
    // 註解中添加tag = true,表示當前字段內容爲tag內容
    @Column(name = "module", tag = true)
    private String module;
    @Column(name = "level", tag = true)
    private String level;
    @Column(name = "device_id", tag = true)
    private String deviceId;
    @Column(name = "msg")
    private String msg;
}

保存數據

如下代碼爲單條日誌保存,influxdb-java亦支持批量保存(由於與InfluxDB通信均是經過http,所以建議批量保存以減小性能損耗).

LogInfo logInfo = LogInfo.builder()
        .level(jsonObject.getString("level"))
        .module(module)
        .deviceId(deviceId)
        .msg(jsonObject.getString("msg"))
        .build();
    Point point = Point.measurementByPOJO(logInfo.getClass())
        .addFieldsFromPOJO(logInfo)
        .time(jsonObject.getLong("time"), TimeUnit.MILLISECONDS)
        .build();
    // 出於業務考量,設備能夠設置不一樣的保存策略(策略名爲固定前綴+設備ID)
    influxDB.write(influxDBUtils.database, InfluxDbUtils.policyNamePix + deviceId, point);

查詢數據

由於代碼與業務耦合比較厲害,所以此處僅截選作概要示範.

// InfluxDB支持分頁查詢,所以能夠設置分頁查詢條件
    String pageQuery = " LIMIT " + request.getPageSize() + " OFFSET " + ((request.getPageNum() - 1) * request.getPageSize());
    // 此處查詢全部內容,若是
    String queryCmd = "SELECT * FROM "
        // 查詢指定設備下的日誌信息
        // 要指定從 RetentionPolicyName(保存策略前綴+設備ID).measurement(logInfo) 中查詢指定數據)
        + InfluxDbUtils.policyNamePix + request.getDeviceId() + "." + "logInfo"
        // 添加查詢條件(注意查詢條件選擇tag值,選擇field數值會嚴重拖慢查詢速度)
        + queryCondition
        // 查詢結果須要按照時間排序
        + " ORDER BY time DESC"
        // 添加分頁查詢條件
        + pageQuery;

選擇時序數據庫,不建議使用刪除以及更新操做,所以不作介紹.

能夠經過建立或者RetentionPolicy,來添加或者更新數據的刪除時間.

PS:
若是您以爲個人文章對您有幫助，請關注個人微信公衆號，謝謝!