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/bash4、直接输入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 选项设置为 true7、设置保存策略（多长时间之前的数据需要删除）---默认为 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 truename 策略名称：默认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
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端口安全组）设置账号密码
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/
标签：01,SpringBoot,get,InfluxDB2,InfluxDB1,influxdb,2022,public,influx 来源： https://www.cnblogs.com/Tom-shushu/p/15830776.html