Linux下安装Influxdb1.7.8并整合Grafana与Java

Linux下安装Influxdb1.7.8及Grafana

1.安装

# wget https://dl.influxdata.com/influxdb/releases/influxdb-1.7.8.x86_64.rpm

# sudo yum localinstall influxdb-1.7.8.x86_64.rpm

2.启动

#sudo systemctl start influxdb

配置开机启动

#sudo systemctl enable influxdb

3.防火墙配置

#开放端口

# firewall-cmd --add-port=8086/tcp --permanent

这里解释一下,如果在开放端口的时候报错,先看一下是否开启了防火墙

通过systemctl status firewalld查看firewalld状态。

通过systemctl start firewalld开启防火墙,没有任何提示即开启成功。

再次通过systemctl status firewalld查看firewalld状态。

如果要关闭防火墙设置,过systemctl stop firewalld这条指令来关闭该功能。

4.启动cli命令

# influx

5.Influxdb CLI实操

查看数据库:

show databases

创建数据库:

create database tableName

删除数据库:

drop databese tableName

使用数据库:

use tableName

查看表:

show measurements

删除表:

drop measurement 表名

创建用户:

create user influx with password ‘influxdb’ (创建一个普通用户,密码为‘influxdb’)

create user “admin” with password ‘admin’ with all privileges(创建一个管理员用户,给他所有权限)

查看用户

show users

修改密码

set password for influx = ‘influx’

删除用户

drop user admin

数据库访问权限

用户授权

grant all privileges to influx(授权管理员权限)

撤销授权

revoke all privileges from influx

数据保留策略:

常用函数:

1.count()函数

返回一个(field)字段中的非空值的数量。

2.DISTINCT()函数

返回一个字段(field)的唯一值。

3.MEAN() 函数

返回一个字段(field)中的值的算术平均值(平均值)。字段类型必须是长整型或float64

4.MEDIAN()函数

从单个字段(field)中的排序值返回中间值(中位数)。字段值的类型必须是长整型或float64格式。

5.SPREAD()函数

返回字段的最小值和最大值之间的差值。数据的类型必须是长整型或float64。

6.SUM()函数

返回一个字段中的所有值的和。字段的类型必须是长整型或float64。

7.BOTTOM()函数

作用:返回一个字段中最小的N个值。字段类型必须是长整型或float64类型。

8.FIRST()函数

作用:返回一个字段中最老的取值。

9.LAST()函数

作用:返回一个字段中最新的取值。

10.MAX()函数

作用:返回一个字段中的最大值。该字段类型必须是长整型,float64,或布尔类型。

11.MIN()函数

作用:返回一个字段中的最小值。该字段类型必须是长整型,float64,或布尔类型。

12.PERCENTILE()函数

作用:返回排序值排位为N的百分值。字段的类型必须是长整型或float64。

13.DERIVATIVE()函数

作用:返回一个字段在一个series中的变化率。(这个函数还是看官网舒服点,反正我没怎么看懂)

14.DIFFERENCE()函数

作用:返回一个字段中连续的时间值之间的差异。字段类型必须是长整型或float64。

15.ELAPSED()函数

作用:返回一个字段在连续的时间间隔间的差异,间隔单位可选,默认为1纳秒。

16.MOVING_AVERAGE()函数

作用:返回一个连续字段值的移动平均值,字段类型必须是长整形或者float64类型。

17.NON_NEGATIVE_DERIVATIVE()函数

作用:返回在一个series中的一个字段中值的变化的非负速率。

18.STDDEV()函数

作用:返回一个字段中的值的标准偏差。值的类型必须是长整型或float64类型。

 

 

Grafana:可视化图形界面

Linux下安装Grafana

官网:https://grafana.com/

安装

wget https://dl.grafana.com/enterprise/release/grafana-enterprise-8.1.3-1.x86_64.rpm

sudo yum install grafana-enterprise-8.1.3-1.x86_64.rpm

启动

sudo systemctl daemon-reload
sudo systemctl start grafana-server
sudo systemctl status grafana-server

配置开机自启

sudo systemctl enable grafana-server

打开防火墙访问3000端口

账号密码为admin

与Influxdb整合(好多图片没有加载出来,这里就写一下步骤了)

1.配置数据源

2.新建dashboards

数据配置

查看配置好的数据

与Java整合

1、pom.xml引入相关jar文件,如下:

<!-- 引入influxdb依赖 -->
 <dependency>
 <groupId>org.influxdb</groupId>
 <artifactId>influxdb-java</artifactId>
 <version>2.8</version>
 </dependency>
2、influxDB工具类封装:

package com.mt.core.util;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDB.ConsistencyLevel;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Point.Builder;
import org.influxdb.dto.Pong;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import lombok.Data;
​
/**
 * InfluxDB数据库连接操作类
 *
 * @author Simon
 */
​
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",
 "default", 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) {
 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 database
 * 数据库
 * @param retentionPolicy
 * 保存策略
 * @param consistency
 * 一致性
 * @param records
 * 要保存的数据(调用BatchPoints.lineProtocol()可得到一条record)
 */
 public void batchInsert(final String database, final String retentionPolicy, final 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;
 }
​
}
3、查询数据

InfluxDB支持一次查询多个SQL,SQL之间用逗号隔开即可。

public static void main(String[] args) {
 InfluxDBConnection influxDBConnection = new InfluxDBConnection("root", "Password01", "localhost", "devops", "tk_test");
 QueryResult results = influxDBConnection
 .query("SELECT * FROM mt order by time desc limit 1000");
 //results.getResults()是同时查询多条SQL语句的返回值,此处我们只有一条SQL,所以只取第一个结果集即可。
 Result oneResult = results.getResults().get(0);
 if (oneResult.getSeries() != null) {
 List<List<Object>> valueList = oneResult.getSeries().stream().map(Series::getValues)
 .collect(Collectors.toList()).get(0);
 if (valueList != null && valueList.size() > 0) {
 for (List<Object> value : valueList) {
 Map<String, String> map = new HashMap<String, String>();
 // 数据库中字段1取值
 String field1 = value.get(0) == null ? null : value.get(0).toString();
 // 数据库中字段2取值
 String field2 = value.get(1) == null ? null : value.get(1).toString();
 // TODO 用取出的字段做你自己的业务逻辑……
 }
 }
 }
 }
4、插入数据

InfluxDB的字段类型,由第一条插入的值得类型决定;tags的类型只能是String型,可以作为索引,提高检索速度。

public static void main(String[] args) {
 InfluxDBConnection influxDBConnection = new InfluxDBConnection("root", "Password01", "localhost", "devops", "tk_test");
 Map<String, String> tags = new HashMap<String, String>();
 tags.put("tag1", "标签值");
 Map<String, Object> fields = new HashMap<String, Object>();
 fields.put("field1", "String类型");
 // 数值型,InfluxDB的字段类型,由第一天插入的值得类型决定
 fields.put("field2", 3.141592657);
 // 时间使用毫秒为单位
 influxDBConnection.insert("表名", tags, fields, System.currentTimeMillis(), TimeUnit.MILLISECONDS);
 }
 

 

 

上一篇:Grafana


下一篇:5分钟搞定Loki告警多渠道接入