VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > c#编程 >
  • Docker安装InfluxDB1.x和InfluxDB2.x以及与SpringBoot整合

两者区别:

复制代码
1.x 版本使用 influxQL 查询语言,2.x 和 1.8+(beta) 使用 flux 查询语法;相比V1 移除了database 和 RP,增加了bucket。
V2具有以下几个概念:
timestamp、field key、field value、field set、tag key、tag value、tag set、measurement、series、point、bucket、bucket schema、organization
新增的概念:
bucket:所有 InfluxDB 数据都存储在一个存储桶中。一个桶结合了数据库的概念和存储周期(时间每个数据点仍然存在持续时间)。一个桶属于一个组织
bucket schema:具有明确的schema-type的存储桶需要为每个度量指定显式架构。测量包含标签、字段和时间戳。显式模式限制了可以写入该度量的数据的形状。
organization:InfluxDB组织是一组用户的工作区。所有仪表板、任务、存储桶和用户都属于一个组织。
复制代码
本文所有代码:https://github.com/Tom-shushu/InfluxDB1.xAnd2.x-SpringBoot
新得阅读地址:http://www.zhouhong.icu/post/161

 

一、InfluxDB1.x Docker安装以及与Boot整合

A、docker安装InfluxDB1.x (influxdb1.8.4)
1、安装:
docker run -d  --name influxdb -p 8086:8086 influxdb:1.8.4 
2、查看

 

docker ps -a
3、进入docker的influx中
docker exec -it daf88772adc9 /bin/bash
4、直接输入influx启动
influx
5、修改账户密码
复制代码
# 显示用户
SHOW USERS
# 创建用户
CREATE USER "username" WITH PASSWORD 'password'
# 赋予用户管理员权限
GRANT ALL PRIVILEGES TO username
# 创建管理员权限的用户
CREATE USER <username> WITH PASSWORD '<password>' WITH ALL PRIVILEGES
# 修改用户密码
SET PASSWORD FOR username = 'password'
# 撤消权限
REVOKE ALL ON mydb FROM username
# 查看权限
SHOW GRANTS FOR username
# 删除用户
DROP USER "username"
复制代码

 

6、在配置文件启用认证
默认情况下,influxdb的配置文件是禁用认证策略的,所以需要修改设置一下。
编辑配置文件vim /etc/influxdb/influxdb.conf,把 [http] 下的 auth-enabled 选项设置为 true
7、设置保存策略(多长时间之前的数据需要删除)---默认为 autogen 永久不删除
a、查看数据库的保存策略
show retention policies on 数据库名
例子:
复制代码
# 选择使用telegraf数据库
> use influx_test;
Using database influx_test
# 查询数据保存策略
> show retention policies on influx_test
name    duration shardGroupDuration replicaN default
----    -------- ------------------ -------- -------
autogen 0s       168h0m0s           1        true
复制代码

 

name 策略名称:默认autogen
duration 持续时间: 0s 代表无限制
shardGroupDuration shardGroup数据存储时间:shardGroup是InfluxDB的一个基本存储结构, 应该大于这个时间的数据在查询效率上应该有所降低。
replicaN 副本个数:1 代表只有一个副本
default 是否默认策略:true 代表设置为该数据库的默认策略
b、设置保存策略
# 新建一个策略
CREATE RETENTION POLICY "策略名称" ON 数据库名 DURATION 时长 REPLICATION 副本个数;
# 新建一个策略并且直接设置为默认策略
CREATE RETENTION POLICY "策略名称" ON 数据库名 DURATION 时长 REPLICATION 副本个数 DEFAULT;
例子:
# 创建新的默认策略role_01保留数据时长1小时
> CREATE RETENTION POLICY "1hour" ON influx_test DURATION 1h REPLICATION 1 DEFAULT;
c、修改保存策略

 

ALTER RETENTION POLICY "策略名称" ON "数据库名" DURATION 时长
ALTER RETENTION POLICY "策略名称" ON "数据库名" DURATION 时长 DEFAULT
d、删除保存策略

 

drop retention POLICY "策略名" ON "数据库名"
8、使用桌面可视化工具连接数据库
工具链接:https://github.com/CymaticLabs/InfluxDBStudio/releases/download/v0.2.0-beta.1/InfluxDBStudio-0.2.0.zip

 

如果刚才没有设置密码,这里可以不需要填写密码,如果有账号密码则需要勾上下面的Use SSL
连接成功后如下:

 

B、InfluxDB1.x与Spring整合(只列举部分代码,后面会放上整个项目的GitHub地址)
整个项目结构如下:

 

1、引入依赖 (其他依赖未显示全,后面会放上整个项目的GitHub地址)

 

复制代码
<dependency>
    <groupId>com.influxdb</groupId>
    <artifactId>influxdb-client-java</artifactId>
    <version>4.0.0</version>
</dependency>
<dependency>
    <groupId>org.influxdb</groupId>
    <artifactId>influxdb-java</artifactId>
    <version>2.20</version>
</dependency>
复制代码
2、新建yml文件

 

influx:
  url: 'http://xxx.xx.xxx.xx:8086'
  password: 'password'
  username: 'username'
3、连接配置 InfluxDBConfig

 

复制代码
@Data
@Configuration
@ConfigurationProperties(prefix = "influx")
public class InfluxDBConfig {
    private String url;
    private String username;
    private String password;
    /**
      * description: 用于查询
      * date: 2022/1/20 23:11
      * author: zhouhong
      * @param  * @param null
      * @return
      */
    @Bean(destroyMethod = "close")
    public InfluxDB influxDBClient(){
        return InfluxDBFactory.connect(this.url, this.username, this.password);
    }
    /**
      * description: 用于写入
      * date: 2022/1/20 23:12
      * author: zhouhong
      * @param  * @param null
      * @return
      */
    @Bean(name = "influxDbWriteApi",destroyMethod = "close")
    public WriteApi influxDbWriteApi(){
        InfluxDBClient influxDBClient = InfluxDBClientFactory.createV1(this.url, this.username,
                this.password.toCharArray(), "influx_test", "autogen");
        return influxDBClient.getWriteApi();
    }
}
复制代码
4、封装用于查询的方法

 

复制代码
@Component
public class InfluxUtil {
    /**
      * description: 通用查询
      * date: 2022/1/20 23:13
      * author: zhouhong
      * @param  * @param null
      * @return
      */
    public QueryResult query(String command, String database, InfluxDB influxDB) {
        Query query = new Query(command, database);
        return influxDB.query(query);
    }
}
复制代码
5、新建需要写入的数据的实体类、需要返回的类(省略,具体参考github示例)InsertParams.java InfluxResult.java
6、新建server层和impl实现类
InfluxServiceImpl.java 如下:

 

复制代码
/**
  * description: 时序数据库Impl
  * date: 2022/1/16 20:47
  * author: zhouhong
  */
@Service
@Slf4j
public class InfluxServiceImpl implements InfluxService {
    @Resource(name = "influxDbWriteApi")
    private WriteApi influxDbWriteApi;
    @Resource(name = "influxDBClient")
    private InfluxDB influxDBClient;
    @Autowired
    private InfluxUtil influxUtil;
    @Override
    public void insert(InsertParams insertParams) {
        influxDbWriteApi.writeMeasurement(WritePrecision.MS, insertParams);
    }
    @Override
    public Object queryAll(InsertParams insertParams) {
        List<InfluxResult> list = new ArrayList<>();
        InfluxResult influxResult = new InfluxResult();
        String sql = "SELECT * FROM \"influx_test\" WHERE time > '2022-01-16'  tz('Asia/Shanghai')";
        QueryResult queryResult = influxUtil.query(sql, "influx_test", influxDBClient);
        queryResult.getResults().get(0).getSeries().get(0).getValues().forEach(item -> {
            influxResult.setTime(item.get(0).toString());
            influxResult.setCurrent(item.get(1).toString());
            influxResult.setEnergyUsed(item.get(2).toString());
            influxResult.setPower(item.get(3).toString());
            influxResult.setVoltage(item.get(4).toString());
            list.add(influxResult);
        });
        return list;
    }
    @Override
    public Object querySumByOneDay(InsertParams insertParams) {
        String sql = "SELECT  SUM(voltage)  FROM \"influx_test\" WHERE time > '2022-01-18'  GROUP BY time(1d)  tz('Asia/Shanghai')";
        QueryResult queryResult = influxUtil.query(sql, "influx_test", influxDBClient);
        return queryResult.getResults().get(0).getSeries().get(0);
    }
}
复制代码
7、controller层 InfluxDbController.java(返回结果是封装过后的,详情见github示例)

 

复制代码
@RestController
public class InfluxDbController {
    @Autowired
    private InfluxService influxService;
    /**
      * description: 时序数据库插入测试
      * date: 2022/1/16 23:00
      * author: zhouhong
      * @param  * @param null
      * @return
      */
    @PostMapping("/influxdb/insert")
    public ResponseData insert(@RequestBody InsertParams insertParams) {
        influxService.insert(insertParams);
        return new SuccessResponseData();
    }
    /**
     * description: 时序数据库查询全部数据测试
     * date: 2022/1/16 23:00
     * author: zhouhong
     * @param  * @param null
     * @return
     */
    @PostMapping("/influxdb/queryAll")
    public ResponseData query(@RequestBody InsertParams insertParams) {
        return new SuccessResponseData(influxService.queryAll(insertParams));
    }
    /**
     * description: 时序数据库按天查询当前电压总和测试
     * date: 2022/1/16 23:00
     * author: zhouhong
     * @param  * @param null
     * @return
     */
    @PostMapping("/influxdb/queryByOneDay")
    public ResponseData queryByOneDay(@RequestBody InsertParams insertParams) {
        return new SuccessResponseData(influxService.querySumByOneDay(insertParams));
    }
}
复制代码
8、PostMan测试(注意需要先新建一个 数据库---influx_test)
8.1 插入测试 localhost:9998/influxdb/insert
入参:

 

{
    "energyUsed":243.78,
    "power":54.50,
    "current":783.34,
    "voltage":44.09
}
返回:

 

复制代码
{
    "success": true,
    "code": 200,
    "message": "请求成功",
    "localizedMsg": "请求成功",
    "data": null
}
复制代码
8.2、查询全部(注意,这里返回结果我封装了一下)localhost:9998/influxdb/queryAll
入参:

 

{
}
返回:

 

复制代码
{
    "success": true,
    "code": 200,
    "message": "请求成功",
    "localizedMsg": "请求成功",
    "data": [
        {
            "energyUsed": "243.78",
            "power": "54.5",
            "current": "783.34",
            "voltage": "44.09",
            "time": "2022-01-20T23:44:00.626+08:00"
        },
        {
            "energyUsed": "243.78",
            "power": "54.5",
            "current": "783.34",
            "voltage": "44.09",
            "time": "2022-01-20T23:44:00.626+08:00"
        }
    ]
}
复制代码
8.3聚合查询(统计2022-01-18到现在,以天为单位每天的用电量之和) localhost:9998/influxdb/queryByOneDay 精度问题暂时没处理
入参:
{ }
返回:

 

复制代码
{
    "success": true,
    "code": 200,
    "message": "请求成功",
    "localizedMsg": "请求成功",
    "data": {
        "name": "influx_test",
        "tags": null,
        "columns": [
            "time",
            "sum"
        ],
        "values": [
            [
                "2022-01-18T00:00:00+08:00",
                null
            ],
            [
                "2022-01-19T00:00:00+08:00",
                null
            ],
            [
                "2022-01-20T00:00:00+08:00",
                481.07000000000005
            ]
        ]
    }
}
复制代码
C、常见的查询SQL 后面加上 tz('Asia/Shanghai') 解决时区差
1、查所指定时间之后的所有

 

SELECT * FROM "real_water_amount" where time  > '2022-01-01' tz('Asia/Shanghai')

2、查询平均值 mean()

 

SELECT mean(value) FROM "real_water_amount" where time  > '2022-01-01' tz('Asia/Shanghai')

3、查询最大最小值 max() min()

 

SELECT max(value) FROM "real_water_amount" where time  > '2022-01-01' tz('Asia/Shanghai')

4、按年、月、天、周、小时、分钟、秒统计

 

SELECT sum(value) FROM "real_water_amount" where time  > '2022-01-01'  group by time(1d)  tz('Asia/Shanghai')

5、按照列过滤

 

SELECT * FROM "real_water_amount" where time  > '2022-01-01'   and  iotId = '8ecJY59UJd1jwPLBmJA5000000'

二、InfluxDB2.x Docker安装以及与Boot整合

A、Docker安装InfluxDB2.x
1、安装:默认拉取最新版本
docker run -d --name influxdb -p 8086:8086 influxdb
2、查看
docker ps -a
3、浏览器访问 IP:8086 (注意:部署在远程服务器上需要开启8086端口安全组)设置账号密码

 

从上到下为:账号(zhouhong)、密码(66668888)、确认密码(66668888)、组织(my_influxdb)、Buucket(Tom);完了之后点击 Quick Start
4、然后点击 Data -- > Buucket 就可以看到我们刚才创建的 名字为 Tom 的 Buucket了

5、点击 API Tokens 获取当前用户的 Token(整合时需要)

 

6、设置Bucket的保存策略

 

 

准备工作完成,开始整合
B、InfluxDB2.x与SpringBoot整合
1、依赖

 

复制代码
<dependency>
    <groupId>com.influxdb</groupId>
    <artifactId>influxdb-client-java</artifactId>
    <version>4.0.0</version>
</dependency>
<dependency>
    <groupId>org.influxdb</groupId>
    <artifactId>influxdb-java</artifactId>
    <version>2.20</version>
</dependency>
复制代码
2、yml配置文件

 

influx:
  influxUrl: 'http://XXX.XX.XXX.XX:8086'
  bucket: 'tom'
  org: 'my_influxdb'
  token: 'Rt23UemGI_cfS-lFDrurtjh46P1enfhrji-KrZYR04wUR1Yxw_oBCZPL6GmFYSDn20Q9gM_P9DIBhHc2RJjNkA=='
3、配置类

 

复制代码
@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.NS, object);
        }
    }
    /**
     * 读取数据
     */
    public List<FluxTable> queryTable(String fluxQuery){
        return client.getQueryApi().query(fluxQuery, org);
    }
}
复制代码
复制代码
@Data
@Configuration
@ConfigurationProperties(prefix = "influx")
public class InfluxConfig {
    /**
     * url地址
     */
    private String influxUrl;
    /**
     * 桶(表)
     */
    private String bucket;
    /**
     * 组织
     */
    private String org;
    /**
     * token
     */
    private String token;
    /**
     * 初始化bean
     */
    @Bean(name = "influx")
    public InfluxBean InfluxBean() {
        return new InfluxBean(influxUrl, bucket, org, token);
    }
}
复制代码
4、实现类

 

复制代码
@Service
@Slf4j
public class InfluxServiceImpl implements InfluxService {
    @Resource
    private InfluxBean influxBean;
    @Override
    public void insert(InsertParams insertParams) {
        insertParams.setTime(Instant.now());
        influxBean.write(insertParams);
    }
    @Override
    public List<InfluxResult> queue(){
        // 下面两个 private 方法 赋值给 list 查询对应的数据
        List<FluxTable> list = queryInfluxAll();
        List<InfluxResult> results = new ArrayList<>();
        for (int i = 0; i < list.size(); i++) {
            for (int j = 0; j < list.get(i).getRecords().size(); j++) {
                InfluxResult influxResult = new InfluxResult();
                influxResult.setCurrent(list.get(i).getRecords().get(j).getValues().get("current").toString());
                influxResult.setEnergyUsed(list.get(i).getRecords().get(j).getValues().get("energyUsed").toString());
                influxResult.setPower(list.get(i).getRecords().get(j).getValues().get("power").toString());
                influxResult.setVoltage(list.get(i).getRecords().get(j).getValues().get("voltage").toString());
                influxResult.setTime(list.get(i).getRecords().get(j).getValues().get("_time").toString());
                System.err.println(list.get(i).getRecords().get(j).getValues().toString());
                results.add(influxResult);
            }
        }
        return results;
    }
    /**
      * description: 查询一小时内的InsertParams所有数据
      * date: 2022/1/21 13:44
      * author: zhouhong
      * @param  * @param null
      * @return
      */
    private List<FluxTable> queryInfluxAll(){
        String query = " from(bucket: \"tom\")" +
                "  |> range(start: -60m, stop: now())" +
                "  |> filter(fn: (r) => r[\"_measurement\"] == \"influx_test\")" +
                "  |> pivot( rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\" )";
        return influxBean.queryTable(query);
    }
    /**
      * description: 根据某一个字段的值过滤(查询 用电量 energyUsed 为 322 的那条记录)
      * date: 2022/1/21 12:44
      * author: zhouhong
      * @param  * @param null
      * @return
      */
    public List<FluxTable> queryFilterByEnergyUsed(){
        String query = " from(bucket: \"tom\")" +
                "  |> range(start: -60m, stop: now())" +
                "  |> filter(fn: (r) => r[\"_measurement\"] == \"influx_test\")" +
                "  |> filter(fn: (r) => r[\"energyUsed\"] == \"322\")" +
                "  |> pivot( rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\" )";
        return influxBean.queryTable(query);
    }
}
复制代码
C、测试
1、插入 localhost:9998/inlfuxdb/insert
入参:

 

{
    "energyUsed":"23.12",
    "power":"321.60",
    "current":"782.72",
    "voltage":"67.43"
}
返回:
复制代码
{
    "success": true,
    "code": 200,
    "message": "请求成功",
    "localizedMsg": "请求成功",
    "data": null
}
复制代码
2、查询所有
入参:
{}
返回:

 

复制代码
{
    "success": true,
    "code": 200,
    "message": "请求成功",
    "localizedMsg": "请求成功",
    "data": [
        {
            "energyUsed": "23.12",
            "power": "321.60",
            "current": "782.72",
            "voltage": "67.43",
            "time": "2022-01-20T17:51:01.819Z"
        },
        {
            "energyUsed": "243.78",
            "power": "541.50",
            "current": "32.34",
            "voltage": "89.09",
            "time": "2022-01-20T17:33:47.246Z"
        }
    ]
}
复制代码
D、Flux常见查询语句
1、指定数据源:from(bucket:"tom")
指定时间范围:
使用管道转发运算符 ( |>) 将数据从数据源通过管道传输到range() 函数,该函数指定查询的时间范围。它接受两个参数:start和stop。范围可以是使用相对负持续时间 或使用绝对时间

 

//使用绝对时间
from(bucket:"tom")
  |> range(start: 2022-01-05T23:30:00Z, stop: 2022-01-21T00:00:00Z) 
//过去十五天的数据
from(bucket:"tom")
  |> range(start: -15d)

2、数据过滤
将范围数据传递到filter()函数中,以根据数据属性或列缩小结果范围
复制代码
// 根据 _measurement 和 _field 过滤 
 from(bucket:"tom")
  |> range(start: -15d)
  |> filter(fn: (r) =>
    r._measurement == "influx_test" and
    r._field == "power" and
    r.energyUsed == "23.12"
  )
复制代码

 

3、数据转换
使用函数,将数据聚合为平均值、下采样数据等
复制代码
from(bucket:"tom")
  |> range(start: -15d)
  |> filter(fn: (r) =>
    r._measurement == "influx_test"
  )
  |> window(every: 10m)
  
  from(bucket:"tom")
  |> range(start: -15d)
  |> filter(fn: (r) =>
    r._measurement == "influx_test"
  )
  |> window(every: 10m)
  |> mean()
复制代码
其他查询函数请查看官网:https://docs.influxdata.com/flux/v0.x/stdlib/universe/

来源:
https://www.cnblogs.com/Tom-shushu/p/15830776.html

相关教程