InfluxDB工具类
1 package com.influxdb.test;
2
3 import java.util.Map;
4
5 import org.influxdb.InfluxDB;
6 import org.influxdb.InfluxDBFactory;
7 import org.influxdb.dto.Point;
8 import org.influxdb.dto.Point.Builder;
9 import org.influxdb.dto.Query;
10 import org.influxdb.dto.QueryResult;
11 import org.springframework.util.StringUtils;
12
13 /**
14 * 时序数据库 InfluxDB 连接
15 * @author Dai_LW
16 *
17 */
18 public class InfluxDBConnect {
19
20 private String username;//用户名
21 private String password;//密码
22 private String openurl;//连接地址
23 private String database;//数据库
24
25 private InfluxDB influxDB;
26
27
28 public InfluxDBConnect(String username, String password, String openurl, String database){
29 this.username = username;
30 this.password = password;
31 this.openurl = openurl;
32 this.database = database;
33 }
34
35 /**连接时序数据库；获得InfluxDB**/
36 public InfluxDB influxDbBuild(){
37 if(influxDB == null){
38 influxDB = InfluxDBFactory.connect(openurl, username, password);
39 influxDB.createDatabase(database);
40
41 }
42 return influxDB;
43 }
44
45 /**
46 * 设置数据保存策略
47 * defalut 策略名 /database 数据库名/ 30d 数据保存时限30天/ 1 副本个数为1/ 结尾DEFAULT 表示 设为默认的策略
48 */
49 public void createRetentionPolicy(){
50 String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
51 "defalut", database, "30d", 1);
52 this.query(command);
53 }
54
55 /**
56 * 查询
57 * @param command 查询语句
58 * @return
59 */
60 public QueryResult query(String command){
61 return influxDB.query(new Query(command, database));
62 }
63
64 /**
65 * 插入
66 * @param measurement 表
67 * @param tags 标签
68 * @param fields 字段
69 */
70 public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields){
71 Builder builder = Point.measurement(measurement);
72 builder.tag(tags);
73 builder.fields(fields);
74
75 influxDB.write(database, "", builder.build());
76 }
77
78 /**
79 * 删除
80 * @param command 删除语句
81 * @return 返回错误信息
82 */
83 public String deleteMeasurementData(String command){
84 QueryResult result = influxDB.query(new Query(command, database));
85 return result.getError();
86 }
87
88 /**
89 * 创建数据库
90 * @param dbName
91 */
92 public void createDB(String dbName){
93 influxDB.createDatabase(dbName);
94 }
95
96 /**
97 * 删除数据库
98 * @param dbName
99 */
100 public void deleteDB(String dbName){
101 influxDB.deleteDatabase(dbName);
102 }
103
104 public String getUsername() {
105 return username;
106 }
107
108 public void setUsername(String username) {
109 this.username = username;
110 }
111
112 public String getPassword() {
113 return password;
114 }
115
116 public void setPassword(String password) {
117 this.password = password;
118 }
119
120 public String getOpenurl() {
121 return openurl;
122 }
123
124 public void setOpenurl(String openurl) {
125 this.openurl = openurl;
126 }
127
128 public void setDatabase(String database) {
129 this.database = database;
130 }
131 }
测试类
1 package com.influxdb.test;
2
3 import java.util.ArrayList;
4 import java.util.Date;
5 import java.util.HashMap;
6 import java.util.List;
7 import java.util.Map;
8
9 import junit.framework.Assert;
10
11 import org.influxdb.dto.QueryResult;
12 import org.influxdb.dto.QueryResult.Result;
13 import org.influxdb.dto.QueryResult.Series;
14 import org.junit.Before;
15 import org.junit.Test;
16 import org.springframework.beans.BeanWrapperImpl;
17
18 import com.influxdb.pojo.CodeInfo;
19
20 public class InfluxDBTest {
21
22 private InfluxDBConnect influxDB;
23 private String username = "admin";//用户名
24 private String password = "admin";//密码
25 private String openurl = "http://127.0.0.1:8086";//连接地址
26 private String database = "test_db";//数据库
27 private String measurement = "sys_code";
28
29 @Before
30 public void setUp(){
31 //创建 连接
32 influxDB = new InfluxDBConnect(username, password, openurl, database);
33
34 influxDB.influxDbBuild();
35
36 influxDB.createRetentionPolicy();
37
38 // influxDB.deleteDB(database);
39 // influxDB.createDB(database);
40 }
41
42 @Test
43 public void testInsert(){//测试数据插入
44 Map<String, String> tags = new HashMap<String, String>();
45 Map<String, Object> fields = new HashMap<String, Object>();
46 List<CodeInfo> list = new ArrayList<CodeInfo>();
47
48 CodeInfo info1 = new CodeInfo();
49 info1.setId(1L);
50 info1.setName("BANKS");
51 info1.setCode("ABC");
52 info1.setDescr("中国农业银行");
53 info1.setDescrE("ABC");
54 info1.setCreatedBy("system");
55 info1.setCreatedAt(new Date().getTime());
56
57 CodeInfo info2 = new CodeInfo();
58 info2.setId(2L);
59 info2.setName("BANKS");
60 info2.setCode("CCB");
61 info2.setDescr("中国建设银行");
62 info2.setDescrE("CCB");
63 info2.setCreatedBy("system");
64 info2.setCreatedAt(new Date().getTime());
65
66 list.add(info1);
67 list.add(info2);
68
69 for(CodeInfo info : list){
70
71 tags.put("TAG_CODE", info.getCode());
72 tags.put("TAG_NAME", info.getName());
73
74 fields.put("ID", info.getId());
75 fields.put("NAME", info.getName());
76 fields.put("CODE", info.getCode());
77 fields.put("DESCR", info.getDescr());
78 fields.put("DESCR_E", info.getDescrE());
79 fields.put("CREATED_BY", info.getCreatedBy());
80 fields.put("CREATED_AT", info.getCreatedAt());
81
82 influxDB.insert(measurement, tags, fields);
83 }
84 }
85
86 @Test
87 public void testQuery(){//测试数据查询
88 String command = "select * from sys_code";
89 QueryResult results = influxDB.query(command);
90
91 if(results.getResults() == null){
92 return;
93 }
94 List<CodeInfo> lists = new ArrayList<CodeInfo>();
95 for (Result result : results.getResults()) {
96
97 List<Series> series= result.getSeries();
98 for (Series serie : series) {
99 // Map<String, String> tags = serie.getTags();
100 List<List<Object>> values = serie.getValues();
101 List<String> columns = serie.getColumns();
102
103 lists.addAll(getQueryData(columns, values));
104 }
105 }
106
107 Assert.assertTrue((!lists.isEmpty()));
108 Assert.assertEquals(2, lists.size());
109 }
110
111 @Test
112 public void testQueryWhere(){//tag 列名 区分大小写
113 String command = "select * from sys_code where TAG_CODE=‘ABC‘";
114 QueryResult results = influxDB.query(command);
115
116 if(results.getResults() == null){
117 return;
118 }
119 List<CodeInfo> lists = new ArrayList<CodeInfo>();
120 for (Result result : results.getResults()) {
121
122 List<Series> series= result.getSeries();
123 for (Series serie : series) {
124 List<List<Object>> values = serie.getValues();
125 List<String> columns = serie.getColumns();
126
127 lists.addAll(getQueryData(columns, values));
128 }
129 }
130
131 Assert.assertTrue((!lists.isEmpty()));
132 Assert.assertEquals(1, lists.size());
133
134 CodeInfo info = lists.get(0);
135
136 Assert.assertEquals(info.getCode(), "ABC");
137
138 }
139
140 @Test
141 public void deletMeasurementData(){
142 String command = "delete from sys_code where TAG_CODE=‘ABC‘";
143 String err = influxDB.deleteMeasurementData(command);
144 Assert.assertNull(err);
145 }
146
147 /***整理列名、行数据***/
148 private List<CodeInfo> getQueryData(List<String> columns, List<List<Object>> values){
149 List<CodeInfo> lists = new ArrayList<CodeInfo>();
150
151 for (List<Object> list : values) {
152 CodeInfo info = new CodeInfo();
153 BeanWrapperImpl bean = new BeanWrapperImpl(info);
154 for(int i=0; i< list.size(); i++){
155
156 String propertyName = setColumns(columns.get(i));//字段名
157 Object value = list.get(i);//相应字段值
158 bean.setPropertyValue(propertyName, value);
159 }
160
161 lists.add(info);
162 }
163
164 return lists;
165 }
166
167 /***转义字段***/
168 private String setColumns(String column){
169 String[] cols = column.split("_");
170 StringBuffer sb = new StringBuffer();
171 for(int i=0; i< cols.length; i++){
172 String col = cols[i].toLowerCase();
173 if(i != 0){
174 String start = col.substring(0, 1).toUpperCase();
175 String end = col.substring(1).toLowerCase();
176 col = start + end;
177 }
178 sb.append(col);
179 }
180 return sb.toString();
181 }
182 }
原文地址：https://www.cnblogs.com/xiaoxiaoMrHuang/p/10724285.html
时间： 2024-11-11 08:43:49