InfluxDb2.0使用与集成

一、场景

        InfluxDb是一个开源时序型数据库，相对于MySql等关系型数据库，InfluxDb更适应于存储日志/监控信息或物联网采集时间相关性信息等场景。

二、安装及使用

        官网: InfluxDB: Purpose-Built Open Source Time Series Database | InfluxData

        下载安装指引:Install InfluxDB | InfluxDB OSS 2.0 Documentation

        指引包括windows、linux、Docker镜像等多种途径的安装

• 1.windows安装示例

在官网中勾选windows版本指引并下载infulxDb压缩包(本文版本2-2.0.8)并解压,压缩包含influx.exe,influxd.exe,README.md文件

使用命令行工具进入解压目录并使用 influx --help 检查influx命令是否正常

使用命令start influxd.exe 启动influxd.exe文件,启动后弹出启动程序(默认端口8086)

访问  http://localhost:8086 进入influxDb用户页面

• 2. InfluxDb用户页面使用

 通过用户页面或CLI初始化信息

查看数据库信息

Sources: InfluxDb与各语言的集成

Buckets: 表信息

Telegraf: 数据收集代理,Use the Telegraf agent to collect and write data | InfluxDB OSS 2.0 Documentation

Scrapers: HTTP(S) 可访问收集数据Create scrapable endpoints for InfluxDB | InfluxDB OSS 2.0 Documentation

Tokens: 访问数据库表token

构建Flux语句执行数据库查询

 三、InfluxDb与SpringBoot集成

在用户界面可以了解如何在java程序中引入InfluxDb并执行插入/查询操作

1. 引入maven依赖包

<dependency>
    <groupId>com.influxdb</groupId>
    <artifactId>influxdb-client-java</artifactId>
    <version>2.0.0</version>
</dependency>

2. 修改spring配置文件参数

spring:
    influx:
        url: http://***.***.***.***:8086
        #bucket可去掉,桶可以在业务上做选择而不写在配置中
        bucket: ***
        org: ***
        token: ***

3. Influx配置类,启动Spring工程时实例化InfluxBean

@Configuration
public class InfluxConfig {
    /**
     * url地址
     */
    @Value("${spring.influx.url}")
    private String influxUrl;
    /**
     * 桶(表)
     */
    @Value("${spring.influx.bucket}")
    private String bucket;
    /**
     * 组织
     */
    @Value("${spring.influx.org}")
    private String org;
    /**
     * token
     */
    @Value("${spring.influx.token}")
    private String token;
    /**
     * 初始化bean
     */
    @Bean(name = "influx")
    public InfluxBean InfluxBean() {
        return new InfluxBean(influxUrl, bucket, org, token);
    }
}

4. Influx实例化类

@Setter
@Getter
public class InfluxBean{
    /**
     * 数据库url地址
     */
    private String influxUrl;
    /**
     * 桶(表)
     */
    private String bucket;
    /**
     * 组织
     */
    private String org;
    /**
     * token
     */
    private String token;
    /**
     * 数据库连接
     */
    private InfluxDBClient client;
    /**
     * 构造方法
     */
    public InfluxBean(String influxUrl, String bucket, String org, String token) {
        this.influxUrl = influxUrl;
        this.bucket = bucket;
        this.org = org;
        this.token = token;
        this.client = getClient();
    }

    /**
     * 获取连接
     */
    private InfluxDBClient getClient() {
        if (client == null) {
            client  = InfluxDBClientFactory.create(influxUrl, token.toCharArray());
        }
        return client;
    }

    /**
     * 写入数据(以秒为时间单位)
     */
    public void write(Object object){
        try (WriteApi writeApi = client.getWriteApi()) {
            writeApi.writeMeasurement(bucket, org, WritePrecision.S, object);
        }
    }

    /**
     * 读取数据
     */
    public List<FluxTable> queryTable(String fluxQuery){
        return client.getQueryApi().query(fluxQuery, org);
    }
}

5. 数据库桶(表)实体类

@Measurement(name = "influxEntity")
@Getter
@Setter
public class InfluxEntity{
    /**
     * 值
     */
    @Column
    Integer value;
    /**
     * 标签字段
     */
    @Column(tag = true)
    String tag;
    /**
     * 时间
     */
    @Column(timestamp = true)
    Instant time;
}

6.在实现类中使用InfluxBean实例及实体类向数据库插入数据

/**
 * 注入InfluxBean
 */
@Resource
private InfluxBean influxBean;

/**
 * 插入InfluxEntity数据
 */
private void writeInfluxEntity(InfluxEntity influxEntity){
    influxBean.write(influxEntity);
}

/**
 * 查询15分钟内的InfluxEntity数据
 */
private List<FluxTable> queryInfluxEntity(){
    String query = 
    " from(bucket: \"influxEntity\")" +
    "  |> range(start: -15m, stop: now())" +
    "  |> filter(fn: (r) => r[\"_measurement\"] == \"influxEntity\")" 
    return influxBean.queryTable(query);
}

/**
 * 业务
 */
public void business(){
    InfluxEntity influxEntity = new InfluxEntity();
    influxEntity.setTime(new Date());
    influxEntity.setTag("tag");
    influxEntity.setValue(1);
    writeInfluxEntity(influxEntity);
    List<FluxTable> queryData = queryInfluxEntity();
}