世界新资讯：springboot接入influxdb

2023-05-30 22:29:14 来源：博客园


(资料图)

转载请注明出处：

1.添加maven依赖
            org.springframework.boot            spring-boot-starter                            org.springframework.boot            spring-boot-starter-test            test                            org.influxdb            influxdb-java            2.14
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;@Servicepublic 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 tags, Map 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 tags, Map 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 tags, Map 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> queryResultProcess(QueryResult queryResult) {        List> mapList = new ArrayList<>();        List resultList =  queryResult.getResults();        //把查询出的结果集转换成对应的实体对象，聚合成list        for(QueryResult.Result query : resultList){            List seriesList = query.getSeries();            if(seriesList != null && seriesList.size() != 0) {                for(QueryResult.Series series : seriesList){                    List columns = series.getColumns();                    String[] keys =  columns.toArray(new String[columns.size()]);                    List> values = series.getValues();                    if(values != null && values.size() != 0) {                        for(List value : values){                            Map 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> list = queryResultProcess(queryResult);        if(list != null && list.size() != 0) {            Map 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 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 records) {        influxDbBuild();        influxDB.write(influxdbDatabase, "", consistency, records);    }}4.进行单元测试验证
@SpringBootTestpublic class InfluxDbTest {    @Autowired    private InfluxDBService influxDBService;    @Test    void contextLoads() {    }    @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);    }}
influxdb 在执行新增测试用例的时候，如果influxdb的数据库中对应的表不存在，会自动创建数据库表
5.查询示例
		  	
      
    
    
    

          

	
      

            
       
      

    

      
        上一篇:
        
金科股份5月30日快速上涨_快播

      
      
        下一篇:
        
金科股份5月30日快速上涨_快播

      
    

    

      
        推荐阅读