Springboot使用influxDB时序数据库的实现

< dependency >

     < groupId >org.influxdb</ groupId >

     < artifactId >influxdb-java</ artifactId >

     < version >2.15</ version >

</ dependency >

spring:

   influx:

     url: *

     password: admin

     user: 123

     database: log_management

@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;

     }

  }

@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;

}

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;