初识influxDB
2020-07-03 本文已影响0人
Tcpieg
influxDB是什么？
参考这篇文章，轻松理解：
InfluxDB（时序数据库），常用的一种使用场景：监控数据统计。每毫秒记录一下电脑内存的使用情况，然后就可以根据统计的数据，利用图形化界面（InfluxDB V1一般配合Grafana）制作内存使用情况的折线图；
可以理解为按时间记录一些数据（常用的监控数据、埋点统计数据等），然后制作图表做统计；
简单来说，就是每一条记录都携带时间。
安装
安装比较简单，进入官网下载，我是windows，安装InfluxDB 1.8版本的windows版本二进制文件。
下载
版本选择
下载完成之后，再安装目录下面解压缩。
解压后文件目录
打开cmd，先运行
influxd.exe
连接成功
不要关闭这个界面，重新打开cmd，运行
influx.exe
操作界面
接下来，操作数据库就好了，参考常用操作命令
使用
参考：入门
这篇文章详细地描述了influxDB的结构，总结起来就是，数据库是database，表是measurement，
一行数据（记录）是point；
point的数据结构由时间戳（time）、标签（tags）、数据（fields）三部分组成。
point属性具体含义
还有一个是保留策略（retention policy），指的是数据的保留时间。基本知道这些就可以开始操作数据库了，需要深入了解的话，可以仔细看看上面的文章。
Java使用
网上有很多操作数据库的代码，我也参考的写了一份。
首先是数据库连接操作类，在网上的资料都是一样的，没什么大问题。
package com.influxdb.influxdbDemo;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.*;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* @program: influxdbDemo
* @description: 数据库连接操作类
* @author: zhudan
* @create: 2020/7/1 14:09
*/
public class InfluxDBConnection {
// 用户名
private String username;
// 密码
private String password;
// 连接地址
private String openurl;
// 数据库
private String database;
// 保留策略
private String retentionPolicy;
private InfluxDB influxDB;
public InfluxDBConnection(String username, String password, String openurl, String database,
String retentionPolicy) {
this.username = username;
this.password = password;
this.openurl = openurl;
this.database = database;
this.retentionPolicy = retentionPolicy == null || retentionPolicy.equals("") ? "autogen" : retentionPolicy;
influxDbBuild();
}
/**
* 创建数据库
*
* @param dbName
*/
@SuppressWarnings("deprecation")
public void createDB(String dbName) {
influxDB.createDatabase(dbName);
}
/**
* 删除数据库
*
* @param dbName
*/
@SuppressWarnings("deprecation")
public void deleteDB(String dbName) {
influxDB.deleteDatabase(dbName);
}
/**
* 测试连接是否正常
*
* @return true 正常
*/
public boolean ping() {
boolean isConnected = false;
Pong pong;
try {
pong = influxDB.ping();
if (pong != null) {
isConnected = true;
}
} catch (Exception e) {
e.printStackTrace();
}
return isConnected;
}
/**
* 连接时序数据库 ，若不存在则创建
*
* @return
*/
public InfluxDB influxDbBuild() {
if (influxDB == null) {
influxDB = InfluxDBFactory.connect(openurl, username, password);
}
try {
// if (!influxDB.databaseExists(database)) {
// influxDB.createDatabase(database);
// }
} catch (Exception e) {
// 该数据库可能设置动态代理，不支持创建数据库
// e.printStackTrace();
} finally {
influxDB.setRetentionPolicy(retentionPolicy);
}
influxDB.setLogLevel(InfluxDB.LogLevel.NONE);
return influxDB;
}
/**
* 创建自定义保留策略
*
* @param policyName 策略名
* @param duration 保存天数
* @param replication 保存副本数量
* @param isDefault 是否设为默认保留策略
*/
public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault) {
String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
database, duration, replication);
if (isDefault) {
sql = sql + " DEFAULT";
}
this.query(sql);
}
/**
* 创建默认的保留策略
*
* @param 策略名：default，操作的数据库，保存天数：30天，保存副本数量：1 设为默认保留策略
*/
public void createDefaultRetentionPolicy() {
String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
"autogen", database, "30d", 1);
this.query(command);
}
/**
* 查询
*
* @param command 查询语句
* @return
*/
public QueryResult query(String command) {
return influxDB.query(new Query(command, database));
}
/**
* 插入
*
* @param measurement 表
* @param tags 标签
* @param fields 字段
*/
public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields, long time,
TimeUnit timeUnit) {
Point.Builder builder = Point.measurement(measurement);
builder.tag(tags);
builder.fields(fields);
if (0 != time) {
builder.time(time, timeUnit);
}
influxDB.write(database, retentionPolicy, builder.build());
}
/**
* 批量写入测点
*
* @param batchPoints
*/
public void batchInsert(BatchPoints batchPoints) {
influxDB.write(batchPoints);
// influxDB.enableGzip();
// influxDB.enableBatch(2000,100,TimeUnit.MILLISECONDS);
// influxDB.disableGzip();
// influxDB.disableBatch();
}
/**
* @param retentionPolicy
* @param points
* @param consistency
* @Description: 批量保存测点
* @return: void
*/
public void bacthInsertPoints(String retentionPolicy, List<Point> points, InfluxDB.ConsistencyLevel consistency) {
BatchPoints.Builder builder = BatchPoints.database(database);
builder.retentionPolicy(retentionPolicy);
for (Point point : points) {
builder.points(point);
}
builder.consistency(consistency);
influxDB.write(builder.build());
}
/**
* 批量写入数据
*
* @param database 数据库
* @param retentionPolicy 保存策略
* @param consistency 一致性
* @param records 要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）
*/
public void batchInsert(final String database, final String retentionPolicy, final InfluxDB.ConsistencyLevel consistency,
final List<String> records) {
influxDB.write(database, retentionPolicy, consistency, records);
}
/**
* 删除
*
* @param command 删除语句
* @return 返回错误信息
*/
public String deleteMeasurementData(String command) {
QueryResult result = influxDB.query(new Query(command, database));
return result.getError();
}
/**
* 关闭数据库
*/
public void close() {
influxDB.close();
}
/**
* 构建Point
*
* @param measurement
* @param time
* @param fields
* @return
*/
public Point pointBuilder(String measurement, long time, Map<String, String> tags, Map<String, Object> fields) {
Point point = Point.measurement(measurement).time(time, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();
return point;
}
/**
* @param measurement
* @param tags
* @param fields
* @Description: 不传时间的构建point
* @return: org.influxdb.dto.Point
*/
public Point pointBuilder(String measurement, Map<String, String> tags, Map<String, Object> fields) {
Point point = Point.measurement(measurement).tag(tags).fields(fields).build();
return point;
}
}
然后测试一下，尝试自己测了一下，问题还是蛮多的。
首先是连接数据库，本来看资料上面ip，在尝试了ip,ip+port,本机（127.0.0.1）+port之后，发现要加上http://之后成功了。
InfluxDBConnection influxDBConnection = new InfluxDBConnection("zhudan", "123456", " http://127.0.0.1:8086/", "test01", "autogen");
System.out.println("---------连接数据库---------\n" + influxDBConnection.ping());
然后开始插入数据试试：
//插入单条记录
String measurement = "test";
Map<String, String> tags = new HashMap<>();
tags.put("dc", "us-west-2");
tags.put("server", "ec2-autodenalicontentpoi-01");
Map<String, Object> fields = new HashMap<>();
fields.put("cpu_usage_user", 5.2);
fields.put("cpu_usage_system", 1.0);
fields.put("cpu_usage_idle", 90.7);
influxDBConnection.insert(measurement, tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
然后查询，但是发现series中tags的值为空的，可是如果查询tag keys又能查出来tags的值（series是一个数据源的唯一标识，series的实际值等于measurement + tags），有大神能告诉我原因的话，感激不尽：
//查询语句
System.out.println("---------查询-----------\n" + influxDBConnection.query("show tag keys from test").getResults());
System.out.println("---------查询-----------\n" + influxDBConnection.query("select * from test002").getResults());
查询结果
web管理界面——chronograf
同样是官网下载
下载chronograf
选择版本
放在安装目录下，解压缩，运行
chronograf.exe
目录
运行界面
然后浏览器打开地址：http://localhost:8888/
首页
输入用户名和密码
连接
继续
因为我没有安装Kapacitor，所以报错，但是不影响进入界面。
进入界面
查看数据库
可视化
