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:
如果您觉得我的文章对您有帮助，可以扫码领取下红包或扫码支持(随意多少,一分钱都是爱)，谢谢！

支付宝红包	支付宝	微信