封装的influxDB工具类

╰+攻爆jí腚メ 2023-06-18 06:52 19阅读 0赞

最近在使用influxDB，但是在网上找了很多工具类都不是很好用。  
大家都知道influxDB查询出来的数据格式不是json格式，比较难以处理，很多网上的工具类都没有对查询的结果进行封装，都是返回一个QueryResult对象。  
遂本文在其基础上进行了一定的修改。

### 1、主要修改的地方： ###

对查询结果进行封装，使用泛型，直接返回javaBean对象。  
说明：本项目使用了springboot，但是大家如果不想用，可以自行稍做修改

### 2、本工具类依赖： ###

<dependency>
                <groupId>org.influxdb</groupId>
                <artifactId>influxdb-java</artifactId>
                <version>2.15</version>
            </dependency>
    		<dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.58</version>
            </dependency>

### 3、工具类 ###

package cc.wllfengshu.utils;
    
    import cc.wellcloud.rtstat.configs.EnvConfig;
    import com.alibaba.fastjson.JSONArray;
    import com.alibaba.fastjson.JSONObject;
    import lombok.extern.slf4j.Slf4j;
    import org.influxdb.InfluxDB;
    import org.influxdb.dto.*;
    import org.influxdb.dto.Point.Builder;
    import org.influxdb.InfluxDBFactory;
    
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.TimeUnit;
    
    /** * 配置influxdb * * @author wangll */
    @Slf4j
    public class InfluxDbUtil { 
    
        public static final String DATABASE = "CMBC_RTSTAT";
        public static final String RETENTION_POLICY = "";
    
        private InfluxDB influxdb = null;
    
        public InfluxDbUtil(EnvConfig envConfig){ 
            if(influxdb == null){ 
                influxdb = InfluxDBFactory.connect(envConfig.getInfluxdbUrl(),envConfig.getInfluxdbUsername(),envConfig.getInfluxdbPassword());
                if (!this.ping()){ 
                    log.error("无法连接到influxDB");
                }
                if (!influxdb.databaseExists(DATABASE)) { 
                    influxdb.enableBatch(2000,10000, TimeUnit.MILLISECONDS);
                    //创建数据库（注意：本文使用默认数据保留策略，如需要修改请自行查资料）
                    influxdb.query(new Query("CREATE DATABASE " + DATABASE));
                    influxdb.setDatabase(DATABASE);
                    influxdb.setLogLevel(InfluxDB.LogLevel.NONE);
                }
                log.info("influxDb初始化完毕");
            }
        }
    
        /** * 查询 * * @param command 查询语句 * @return */
        public <T> List<T> query(String command,Class<T> clazz) { 
            log.info("query command:{}",command);
            JSONArray resultArr = new JSONArray();
            QueryResult query = influxdb.query(new Query(command, DATABASE));
            for (QueryResult.Result result : query.getResults()) { 
                List<QueryResult.Series> series = result.getSeries();
                if(series==null){ 
                    continue;
                }
                for (QueryResult.Series serie : series) { 
                    List<List<Object>> values = serie.getValues();
                    List<String> colums = serie.getColumns();
                    Map<String, String> tags = serie.getTags();
    
                    // 封装查询结果
                    for (int i=0;i<values.size();++i){ 
                        JSONObject jsonData = new JSONObject();
                        if (tags!=null && tags.keySet().size()>0){ 
                            tags.forEach((k,v)-> jsonData.put(k,v));
                        }
                        for (int j=0;j<colums.size();++j){ 
                            jsonData.put(colums.get(j),values.get(i).get(j));
                        }
                        resultArr.add(jsonData);
                    }
                }
            }
            return JSONObject.parseArray(resultArr.toJSONString(), clazz);
        }
    
        /** * 插入 * * @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).tag(tags).fields(fields).time(time, timeUnit);
            influxdb.write(DATABASE, RETENTION_POLICY, builder.build());
        }
    
        /** * 插入 * * @param point */
        public void insert(Point point) { 
            if (null==point){ 
                return;
            }
            influxdb.write(DATABASE, RETENTION_POLICY, point);
        }
    
        /** * 批量写入数据 * * @param points */
        public void batchInsert(List<Point> points) { 
            if (points==null || points.size()<=0){ 
                log.info("数据为空");
                return;
            }
            BatchPoints batchPoints = BatchPoints.database(InfluxDbUtil.DATABASE).retentionPolicy(InfluxDbUtil.RETENTION_POLICY).build();
            for(Point p:points){ 
                batchPoints.point(p);
            }
            influxdb.write(batchPoints);
        }
    
        /** * 删除表 * * @param measurement * @return 返回错误信息 */
        public void deleteMeasurement(String measurement) { 
            log.info("dropMeasurement measurement:{}",measurement);
            influxdb.query(new Query("drop measurement \"" + measurement + "\"", DATABASE));
        }
    
        /** * 删除数据库 */
        public void deleteDb() { 
            influxdb.query(new Query("DROP DATABASE \"" + DATABASE + "\""));
        }
    
        /** * 测试连接是否正常 * * @return true 正常 */
        public boolean ping() { 
            Pong pong = influxdb.ping();
            if (pong != null) { 
                return true;
            }
            return false;
        }
    
        /** * 关闭数据库 */
        public void close() { 
            influxdb.close();
        }
    }

说明：EnvConfig类里面包含influxDB的url/用户名/密码。类如下，如果大家嫌麻烦，可以修改工具类的构造函数即可。EnvConfig类如下：

package cc.wellcloud.rtstat.configs;
    
    import lombok.Data;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.core.annotation.Order;
    import org.springframework.stereotype.Component;
    
    @Slf4j
    @Data
    @Component
    @Order(0)
    public class EnvConfig implements InitializingBean { 
        /** * influxDB配置 */
        @Value("${influxdb_url:http://192.168.40.107:32099}")
        private String influxdbUrl;
        @Value("${influxdb_username:rtstat}")
        private String influxdbUsername;
        @Value("${influxdb_password:rtstat}")
        private String influxdbPassword;
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            log.info("环境变量:{}",this.toString());
        }
    }

### 4、工具类的使用 ###

import cc.wellcloud.rtstat.common.Constant;
    import cc.wellcloud.rtstat.configs.EnvConfig;
    import cc.wellcloud.rtstat.entity.model.AgentData;
    import lombok.extern.slf4j.Slf4j;
    import org.influxdb.dto.Point;
    import org.junit.Before;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class InfluxDbUtilTest { 
    
        @Autowired
        private EnvConfig envConfig;
        private InfluxDbUtil influxDbUtil=null;
    
        @Before
        public void setUp(){ 
            influxDbUtil=new InfluxDbUtil(envConfig);
        }
    
        @Test
        public void ping(){ 
            System.out.println(influxDbUtil.ping());
        }
    
    	@Test
        public void query() { 
            List<AgentData> ads = influxDbUtil.query("select * from " + Constant.INFLUXDB_MEASUREMENT_AGENT_DATA, AgentData.class);
            System.out.println(ads);
        }
    
        @Test
        public void insert1() { 
            //测试插入数据
            Map<String, String> ts = new HashMap<>();
            ts.put("name","张三");
            Map<String, Object> fs = new HashMap<>();
            fs.put("SEX","男");
            fs.put("AGE", 32);
    
            influxDbUtil.insert("test",ts, fs,System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    
        /** * 查看过期时间：SHOW RETENTION POLICIES ON CMBC_RTSTAT * 查看表：show measurements */
        @Test
        public void insert2() { 
            //测试插入数据
            Map<String, String> ts = new HashMap<>();
            ts.put("name","王2");
            Map<String, Object> fs = new HashMap<>();
            fs.put("SEX","女");
            fs.put("AGE", 66);
    
            Point point = Point.measurement("test").time(System.currentTimeMillis(), TimeUnit.MILLISECONDS).tag(ts).fields(fs).build();
            influxDbUtil.insert(point);
        }
    
        @Test
        public void batchInsert() { 
            List<AgentData> as = new ArrayList<>();
            AgentData ad1=new AgentData();
            ad1.setPlatform(4);
            ad1.setAgentNo("wll");
            ad1.setCallLength(999);
            ad1.setCallNum(222);
            ad1.setTalkNum(100);
            as.add(ad1);
            AgentData ad2=new AgentData();
            ad2.setPlatform(5);
            ad2.setAgentNo("liang");
            ad2.setCallLength(888);
            ad2.setCallNum(78);
            as.add(ad2);
    
            List<Point> records = new ArrayList<>();
            for (AgentData a:as) { 
                //1 tags
                Map<String, String> ts = new HashMap<>(6);
                ts.put("platform", String.valueOf(a.getPlatform()));
                ts.put("agentNo",a.getAgentNo());
                //2 fields
                Map<String, Object> fs = new HashMap<>(7);
                fs.put("callLength", a.getCallLength());
                fs.put("callNum",a.getCallNum());
                //3 point
                Point point = Point.measurement(Constant.INFLUXDB_MEASUREMENT_AGENT_DATA).tag(ts).fields(fs).time(System.currentTimeMillis(), TimeUnit.MILLISECONDS).build();
                records.add(point);
            }
            influxDbUtil.batchInsert(records);
        }
    
        @Test
        public void deleteDb() { 
            influxDbUtil.deleteDb();
        }
    
        @Test
        public void deleteMeasurementData() { 
            influxDbUtil.deleteMeasurement(Constant.INFLUXDB_MEASUREMENT_AGENT_DATA);
        }
    
        @After
        public void close(){ 
            influxDbUtil.close();
        }
    }