本文主要是介绍influxdb产生实时数据，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
开发工具：idea 简历maven工程
<dependency><groupId>org.influxdb</groupId><artifactId>influxdb-java</artifactId><version>2.17</version>
</dependency>
<dependency><groupId>com.github.javafaker</groupId><artifactId>javafaker</artifactId><version>1.0.2</version>
</dependency>java工程钟创立两个java文件，目录结构如下
influxDBConnection文件中添加如下代码
package com.influx;import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.*;import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;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) {Point.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 InfluxDB.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;}
}
InfluxDBManager中编写数据写入代码
package com.influx;import com.github.javafaker.Faker;
import com.mysql.cj.x.protobuf.MysqlxDatatypes;
import org.influxdb.InfluxDB;
import org.influxdb.dto.QueryResult;import java.util.HashMap;
import java.util.Locale;
import java.util.concurrent.TimeUnit;public class InfluxDBManager {public static void main(String[] args) throws InterruptedException {//这一行代码数据库名，表名，ip地址，还有持久策略要根据每个人实际情况填写InfluxDBConnection influxDBConnection = new InfluxDBConnection("influx", "influx", "http://192.168.224.129:8086", "influxdb", "influx_retention");Faker faker = new Faker(new Locale("zh-CN"));String[] areas = {"南","北","东","西"};Integer[] altitudes = new Integer[]{500,800,1000,1500};int k = 1;while(true){System.out.println("当前是第轮插入数据:"+ k);int al_intdex = (int)Math.floor(Math.random() * altitudes.length);int ar_index = (int)Math.floor(Math.random() * areas.length);HashMap<String, String> hashMap = new HashMap<String, String>();hashMap.put("altitude", altitudes[al_intdex].toString());hashMap.put("area", areas[ar_index].toString());HashMap<String, Object> stringObjectHashMap = new HashMap<String, Object>();stringObjectHashMap.put("temperature", faker.number().numberBetween(10,30));stringObjectHashMap.put("humidity", faker.number().numberBetween(10,30));influxDBConnection.insert("weather", hashMap, stringObjectHashMap, System.currentTimeMillis(), TimeUnit.MILLISECONDS);Thread.sleep(faker.number().numberBetween(500,1000));k++;}}
}
这篇关于influxdb产生实时数据的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
