InfluxData【付诸实践 02】SpringBoot 集成时序数据库 InfluxDB 应用分享(InfluxDB实例+Feign接口调用InfluxDB API)源码分享

简介: InfluxData【付诸实践 02】SpringBoot 集成时序数据库 InfluxDB 应用分享(InfluxDB实例+Feign接口调用InfluxDB API)源码分享

1.InfluxDB实例

1.1 依赖及配置

<dependency>
  <groupId>org.influxdb</groupId>
  <artifactId>influxdb-java</artifactId>
  <version>2.15</version>
</dependency>
<dependency>
  <groupId>org.projectlombok</groupId>
  <artifactId>lombok</artifactId>
  <optional>true</optional>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-test</artifactId>
  <scope>test</scope>
</dependency>
spring:
  influx:
    url: http://tcloud:18088
    user: admin
    password: admin
    database: mydb
    retentionPolicy: default
    retentionPolicyTime: 30d

1.2 代码实现

配置类:

@Configuration
public class InfluxDatabaseConfig {
    @Value("${spring.influx.url}")
    public String url;
    @Value("${spring.influx.user}")
    public String userName;
    @Value("${spring.influx.password}")
    public String password;
    @Value("${spring.influx.database}")
    public String database;
    @Value("${spring.influx.retentionPolicy}")
    public String retentionPolicy;
    @Value("${spring.influx.retentionPolicyTime}")
    public String retentionPolicyTime;
}

工具类:

@Slf4j
@Component
public class InfluxDatabaseUtil {
    private String database;
    private String retentionPolicy;
    private String retentionPolicyTime;
    private InfluxDB influxdb;
    @Autowired
    private InfluxDatabaseConfig influxDatabaseConfig;
    @PostConstruct
    private void init() {
        // 保存策略
        if (StringUtils.isEmpty(influxDatabaseConfig.retentionPolicy)) {
            this.retentionPolicy = "autogen";
        } else {
            this.retentionPolicy = influxDatabaseConfig.retentionPolicy;
        }
        // 数据保存策略中数据保存时间
        if (StringUtils.isEmpty(influxDatabaseConfig.retentionPolicyTime)) {
            this.retentionPolicyTime = "30d";
        } else {
            this.retentionPolicyTime = influxDatabaseConfig.retentionPolicyTime;
        }
        if (StringUtils.isEmpty(influxDatabaseConfig.database)) {
            // 创建数据库并赋值
            System.out.println("创建数据库!");
        }
        this.database = influxDatabaseConfig.database;
        // 初始化 InfluxDB 实例
        initInfluxDatabase();
        // 数据库设置保存策略
        createRetentionPolicy();
    }
    /**
     * 初始化数据库连接
     */
    private void initInfluxDatabase() {
        influxdb = InfluxDBFactory.connect(influxDatabaseConfig.url, influxDatabaseConfig.userName, influxDatabaseConfig.password);
        influxdb.setDatabase(database);
    }
    /**
     * 设置数据保存策略
     * default 策略名
     * database 数据库名
     * 30d 数据保存时限30天
     * 1 副本个数为1
     * 结尾 DEFAULT 表示 设为默认的策略
     */
    private void createRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                retentionPolicy, database, retentionPolicyTime, 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) {
        Point.Builder builder = Point.measurement(measurement);
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        builder.tag(tags);
        builder.fields(fields);
        log.info("influxDB insert data:[{}]", builder.build().toString());
        influxdb.write(database, "", builder.build());
    }
}

对象封装:

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Usage {
    private String time;
    private String serviceMethod;
    private String userId;
    private Integer count;
    private String url;
}

测试类:

@SpringBootTest
class InfluxDatabaseApplicationTests {
    @Autowired
    private InfluxDatabaseUtil influxDatabaseUtil;
    @Test
    void testInsert() throws InterruptedException {
        Map<String, String> tagsMap = new HashMap<>();
        Map<String, Object> fieldsMap = new HashMap<>();
        System.out.println("influxDB start time :" + System.currentTimeMillis());
        int i = 0;
        do {
            Thread.sleep(100);
            tagsMap.put("user_id", String.valueOf(i % 10));
            tagsMap.put("url", "http://www.baidu.com");
            tagsMap.put("service_method", "testInsert" + (i % 5));
            fieldsMap.put("count", i % 5);
            influxDatabaseUtil.insert("usage", tagsMap, fieldsMap);
            i++;
        } while (i < 50);
    }
    @Test
    void testQuery() {
        QueryResult query = influxDatabaseUtil.query("select * from usage limit 10");
        ArrayList<Object> lists = new ArrayList<>();
        query.getResults().forEach(result -> {
            result.getSeries().forEach(serie -> {
                List<List<Object>> values = serie.getValues();
                List<String> columns = serie.getColumns();
                lists.addAll(getQueryData(columns, values));
            });
        });
        System.out.println(lists.toString());
    }
    private List<Usage> getQueryData(List<String> columns, List<List<Object>> values) {
        List<Usage> lists = new ArrayList<>();
        for (List<Object> list : values) {
            Usage info = new Usage();
            BeanWrapperImpl bean = new BeanWrapperImpl(info);
            for (int i = 0; i < list.size(); i++) {
                String propertyName = setColumns(columns.get(i));
                // 字段名
                Object value = list.get(i);
                // 相应字段值
                bean.setPropertyValue(propertyName, value);
            }
            lists.add(info);
        }
        return lists;
    }
    /*** 转义字段 ***/
    private String setColumns(String column) {
        String[] cols = column.split("_");
        StringBuffer sb = new StringBuffer();
        for (int i = 0; i < cols.length; i++) {
            String col = cols[i].toLowerCase();
            if (i != 0) {
                String start = col.substring(0, 1).toUpperCase();
                String end = col.substring(1).toLowerCase();
                col = start + end;
            }
            sb.append(col);
        }
        return sb.toString();
    }
}

2.Feign接口调用InfluxDB API

2.1 依赖及配置

<dependency>
  <groupId>org.springframework.cloud</groupId>
  <artifactId>spring-cloud-starter-openfeign</artifactId>
  <version>2.2.6.RELEASE</version>
</dependency>
ribbon:
    ReadTimeout: 5000
    ConnectTimeout: 5000

2.2 代码实现

Use the InfluxDB API 文档有详细说明,以下是feign接口:

@FeignClient(name = "influxDbApi", url = "${spring.influx.url}")
public interface InfluxDbApi {
    /**
     * 查询数据库
     *
     * @param q  查询语句
     * @param db 数据库
     * @param u  用户名
     * @param p  密码
     * @return 查询结果
     * @throws Exception 可能出现的异常
     */
    @PostMapping("/query")
    String queryInfo(@RequestParam String db, @RequestParam String u, @RequestParam String p, @RequestParam String q) throws Exception;
}

调用测试类:

@Component
public class ApiTesting {
    @Autowired
    private InfluxDbApi influxDbApi;
    String test() throws Exception {
       /* Map<String, Object> param = new HashMap<>(8);
        param.put("q","CREATE DATABASE testDB");*/
        return influxDbApi.queryInfo("mydb", "admin", "admin", "select * from usage");
    }
}

测试接口:

@RestController
@Slf4j
public class ApiController {
    @Resource(name = "apiTesting")
    private ApiTesting apiTesting;
    @PostMapping("/test")
    public String get() throws Exception {
        log.info("------test------");
        return apiTesting.test();
    }
}
目录
相关文章
|
27天前
|
前端开发 数据可视化 JavaScript
基于React的简易数据可视化图表库集成与应用
基于React的简易数据可视化图表库集成与应用
19 1
|
1月前
|
项目管理 微服务
云效常见问题之将多个微服务应用集成到一次研发流程中发布上线如何解决
云效(CloudEfficiency)是阿里云提供的一套软件研发效能平台,旨在通过工程效能、项目管理、质量保障等工具与服务,帮助企业提高软件研发的效率和质量。本合集是云效使用中可能遇到的一些常见问题及其答案的汇总。
28 0
|
1月前
|
人工智能 Java API
Google Gemini API 接口调用方法
Google 最近发布的 Gemini 1.0 AI 模型通过其升级版,Gemini,标志着公司迄今为止最为强大和多功能的人工智能技术的突破。
|
1月前
|
NoSQL Java 数据库
【问题篇】springboot项目通过数据库限制实例端口号
【问题篇】springboot项目通过数据库限制实例端口号
19 0
|
5天前
|
测试技术 持续交付 开发工具
《Git 简易速速上手小册》第6章:Git 在持续集成/持续部署(CI/CD)中的应用(2024 最新版)
《Git 简易速速上手小册》第6章:Git 在持续集成/持续部署(CI/CD)中的应用(2024 最新版)
29 2
|
1月前
|
缓存 JavaScript 算法
活用 Composition API 核心函数,打造卓越应用(下)
活用 Composition API 核心函数,打造卓越应用(下)
|
1月前
|
人工智能 关系型数据库 Serverless
Serverless 应用引擎常见问题之API生成的函数镜像改为自定义的镜像如何解决
Serverless 应用引擎(Serverless Application Engine, SAE)是一种完全托管的应用平台,它允许开发者无需管理服务器即可构建和部署应用。以下是Serverless 应用引擎使用过程中的一些常见问题及其答案的汇总:
39 3
|
1天前
|
关系型数据库 MySQL 数据库
一台MySQL数据库启动多个实例
一台MySQL数据库启动多个实例
|
1天前
|
存储 SQL 数据库
数据库库表结构设计:原理、实例与最佳实践
数据库库表结构设计:原理、实例与最佳实践
11 0
|
9天前
|
安全 中间件 数据处理
中间件在应用集成
中间件是应用集成的关键,它连接不同系统、平台和应用,解决兼容性问题,实现数据交换和功能互操作。主要应用包括数据集成、服务集成、消息传递、安全与权限管理。选择中间件需考虑兼容性、性能、可扩展性和安全性。中间件简化通信,提高系统性能和可靠性,助力企业实现应用高效协同和商业价值。
14 2