springboot接入influxdb

springboot,接入,influxdb · 浏览次数 : 42

小编点评

**数据库操作类** ```java public class InfluxDBService { @Autowired private InfluxDBClient influxDBClient; public void insert(String measurement, Map tags, Map fields) { influxDBClient.insert(measurement, tags, fields); } } ``` **测试类** ```java @SpringBootTest public class InfluxDbTest { @Autowired private InfluxDBService influxDBService; @Test public void testSave(){ String measurement = "host_cpu_usage_total"; Map tags = new HashMap(); tags.put("host_name", "host2"); tags.put("cpu_core", "core0"); Map fields = new HashMap(); fields.put("cpu_usage", 0.22); fields.put("cpu_idle", 0.56); influxDBService.insert(measurement, tags, fields); } } ``` **测试示例** ```java //测试插入数据 InfluxDBService.insert( "host_cpu_usage_total", tags, fields ); //测试读取数据 List> records = influxDBService.read( "host_cpu_usage_total", tags ); ``` **其他** * 可以使用`@Autowired`注入`InfluxDBService`注入``` * 可以使用`@Test`注解进行测试 * 可以使用`@Mock`注解进行测试 * 可以使用`@Data`注解进行测试

正文

转载请注明出处:

1.添加maven依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.influxdb</groupId>
            <artifactId>influxdb-java</artifactId>
            <version>2.14</version>
        </dependency>

2.spring boot 的application.yaml配置文件中添加influxdb的配置

influxdb:
  url: http://127.0.0.1:8086
  database: influx_db
  username: influx_db_user
  password: influx_db_pwd

  8086 为influxdb默认的端口

3.influxdb配置应用与service方法编写

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
public class InfluxDBService {


    @Value("${influxdb.username}")
    public String influxdbUserName;

    @Value("${influxdb.password}")
    public String influxdbPassword;

    @Value("${influxdb.url}")
    public String influxdbUrl;

    //数据库
    @Value("${influxdb.database}")
    public String influxdbDatabase;

    @PostConstruct
    public void initInfluxDb() {
        this.retentionPolicy = retentionPolicy == null || "".equals(retentionPolicy) ? "autogen" : retentionPolicy;
        this.influxDB = influxDbBuild();
    }
    //保留策略
    private String retentionPolicy;

    private InfluxDB influxDB;

    /**
     * 设置数据保存策略 defalut 策略名 /database 数据库名/ 30d 数据保存时限30天/ 1 副本个数为1/ 结尾DEFAULT
     * 表示 设为默认的策略
     */
    public void createRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                "defalut", influxdbDatabase, "30d", 1);
        this.query(command);
    }

    /**
     * 连接时序数据库;获得InfluxDB
     **/
    private InfluxDB influxDbBuild() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(influxdbUrl, influxdbUserName, influxdbPassword);
            influxDB.setDatabase(influxdbDatabase);
        }
        return influxDB;
    }

    /**
     * 插入
     * @param measurement 表
     * @param tags        标签
     * @param fields      字段
     */
    public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields) {
        influxDbBuild();
        Point.Builder builder = Point.measurement(measurement);
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        builder.tag(tags);
        builder.fields(fields);
        influxDB.write(influxdbDatabase, "", builder.build());
    }

    /**
     * @desc 插入,带时间time
     * @date 2021/3/27
     *@param measurement
     *@param time
     *@param tags
     *@param fields
     * @return void
     */
    public void insert(String measurement, long time, Map<String, String> tags, Map<String, Object> fields) {
        influxDbBuild();
        Point.Builder builder = Point.measurement(measurement);
        builder.time(time, TimeUnit.MILLISECONDS);
        builder.tag(tags);
        builder.fields(fields);
        influxDB.write(influxdbDatabase, "", builder.build());
    }

    /**
     * @desc influxDB开启UDP功能,默认端口:8089,默认数据库:udp,没提供代码传数据库功能接口
     * @date 2021/3/13
     *@param measurement
     *@param time
     *@param tags
     *@param fields
     * @return void
     */
    public void insertUDP(String measurement, long time, Map<String, String> tags, Map<String, Object> fields) {
        influxDbBuild();
        Point.Builder builder = Point.measurement(measurement);
        builder.time(time, TimeUnit.MILLISECONDS);
        builder.tag(tags);
        builder.fields(fields);
        int udpPort = 8089;
        influxDB.write(udpPort,  builder.build());
    }

    /**
     * 查询
     * @param command 查询语句
     * @return
     */
    public QueryResult query(String command) {
        influxDbBuild();
        return influxDB.query(new Query(command, influxdbDatabase));
    }

    /**
     * @desc 查询结果处理
     * @date 2021/5/12
     *@param queryResult
     */
    public List<Map<String, Object>> queryResultProcess(QueryResult queryResult) {
        List<Map<String, Object>> mapList = new ArrayList<>();
        List<QueryResult.Result> resultList =  queryResult.getResults();
        //把查询出的结果集转换成对应的实体对象,聚合成list
        for(QueryResult.Result query : resultList){
            List<QueryResult.Series> seriesList = query.getSeries();
            if(seriesList != null && seriesList.size() != 0) {
                for(QueryResult.Series series : seriesList){
                    List<String> columns = series.getColumns();
                    String[] keys =  columns.toArray(new String[columns.size()]);
                    List<List<Object>> values = series.getValues();
                    if(values != null && values.size() != 0) {
                        for(List<Object> value : values){
                            Map<String, Object> map = new HashMap(keys.length);
                            for (int i = 0; i < keys.length; i++) {
                                map.put(keys[i], value.get(i));
                            }
                            mapList.add(map);
                        }
                    }
                }
            }
        }
        return mapList;
    }

    /**
     * @desc InfluxDB 查询 count总条数
     * @date 2021/4/8
     */
    public long countResultProcess(QueryResult queryResult) {
        long count = 0;
        List<Map<String, Object>> list = queryResultProcess(queryResult);
        if(list != null && list.size() != 0) {
            Map<String, Object> map = list.get(0);
            double num = (Double)map.get("count");
            count = new Double(num).longValue();
        }
        return count;
    }

    /**
     * 查询
     * @param dbName 创建数据库
     * @return
     */
    public void createDB(String dbName) {
        influxDbBuild();
        influxDB.createDatabase(dbName);
    }

    /**
     * 批量写入测点
     *
     * @param batchPoints
     */
    public void batchInsert(BatchPoints batchPoints) {
        influxDbBuild();
        influxDB.write(batchPoints);
    }

    /**
     * 批量写入数据
     *
     * @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) {
        influxDbBuild();
        influxDB.write(database, retentionPolicy, consistency, records);
    }

    /**
     * @desc 批量写入数据
     * @date 2021/3/19
     *@param consistency
     *@param records
     */
    public void batchInsert(final InfluxDB.ConsistencyLevel consistency, final List<String> records) {
        influxDbBuild();
        influxDB.write(influxdbDatabase, "", consistency, records);
    }

}

4.进行单元测试验证

@SpringBootTest
public class InfluxDbTest {

    @Autowired
    private InfluxDBService influxDBService;

    @Test
    void contextLoads() {
    }

    @Test
    public void testSave(){
        String measurement = "host_cpu_usage_total";

        Map<String,String> tags = new HashMap<>();
        tags.put("host_name","host2");
        tags.put("cpu_core","core0");

        Map<String, Object> fields = new HashMap<>();
        fields.put("cpu_usage",0.22);
        fields.put("cpu_idle",0.56);
        influxDBService.insert(measurement, tags, fields);
    }
}

  influxdb 在执行新增测试用例的时候,如果influxdb的数据库中对应的表不存在,会自动创建数据库表

5.查询示例

 

 

 

 

 

  

与springboot接入influxdb相似的内容:

springboot接入influxdb

转载请注明出处: 1.添加maven依赖 org.springframework.boot spring-boot-starter

SpringBoot实战:Spring Boot接入Security权限认证服务

引言 Spring Security 是一个功能强大且高度可定制的身份验证和访问控制的框架,提供了完善的认证机制和方法级的授权功能,是一个非常优秀的权限管理框架。其核心是一组过滤器链,不同的功能经由不同的过滤器。本文将通过一个案例将 Spring Security 整合到 SpringBoot中,要

SpringBoot 接口并发限制(Semaphore)

可以使用 JMeter 辅助测试 https://blog.csdn.net/weixin_45014379/article/details/124190381 @RestController @RequestMapping({"/Test"}) public class test { Logger

SpringBoot实战:轻松实现接口数据脱敏

引言 在现代的互联网应用中,数据安全和隐私保护变得越来越重要。尤其是在接口返回数据时,如何有效地对敏感数据进行脱敏处理,是每个开发者都需要关注的问题。本文将通过一个简单的Spring Boot项目,介绍如何实现接口数据脱敏。 一、接口数据脱敏概述 1.1 接口数据脱敏的定义 接口数据脱敏是指在接口返

[转帖]Springboot 集成 micrometer(actuator/prometheus) 接口报 404

https://blog.csdn.net/qq_26545503/article/details/123313891 原因 主要是spring-boot版本和micrometer版本没有对应上,我用的spring-boot是2.6.0对应的micrometer就是1.8 需要修改两个地方 pom.

使用SpringBoot开发一个POST接口

接着上一篇,开始根据需求开发一个POST接口。

Springboot项目使用Undertow替换内置Tomcat服务器,实现RESTFUL接口web应用

Maven实例:pom.xml文件中添加更换依赖 org.springframework.boot spring-boot-starter

springboot使controller异步调用

调用controller方法,遇到操作时间很长的情况下,不希望前端一直等待操作,而希望前端立马接收到操作启动的反馈,而真正的操作在后端执行,需要用到异步调用的方法。实现步骤如下: 一、配置异步支持: 首先,在应用程序的主类上添加 @EnableAsync 注解,以启用异步支持 import org.

问题排查---应用程序不在接收新请求

问题排查 应用程序不在接收新请求 关键词:springboot,jstack,Arthas 问题描述 查看前端网页,发现所有请求都pending,都超时。但是查看后端程序发现并没有挂掉,cpu,内存都正常。但是日志不打印了。看起来应用程序整体卡死了。 然后重启应用程序,发现又能正常运行了,但是过了半

Springboot中自定义监听器

一、监听器模式图 二、监听器三要素 广播器:用来发布事件 事件:需要被传播的消息 监听器:一个对象对一个事件的发生做出反应,这个对象就是事件监听器 三、监听器的实现方式 1、实现自定义事件 自定义事件需要继承ApplicationEvent类,并添加一个构造函数,用于接收事件源对象。 该事件中添加了