SpringBoot2.0快速整合InfluxDB，新增、批量新增，通过反射机制构建查询结果，简单易懂，拿来即用

网上有不少SpringBoot整合InfluxDB的案例，但是功能比较单一，案例不够完全。这里整理了一个功能全面的案例，包括单一新增、批量新增，通过反射机制对查询结果进行封装，很简单，一看就懂，拿来即用。

这里使用的是SpringBoot 2.1.4.RELEASE，本地需要安装influxdb 1.8.1，接下来看看如何实现的。

第一步，在pom文件中引入influxdb-java架包，还使用了lombok来简化代码量；

<!-- Spring Boot的核心启动器，包含了自动配置、日志和YAML -->
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-web</artifactId>
      </dependency>
      <!-- 测试专用 -->
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-devtools</artifactId>
          <scope>runtime</scope>
          <optional>true</optional>
      </dependency>
      <!-- lombok架包 -->
      <dependency>
          <groupId>org.projectlombok</groupId>
          <artifactId>lombok</artifactId>
          <optional>true</optional>
      </dependency>     
      <!-- influxdb架包 -->    
      <dependency>
          <groupId>org.influxdb</groupId>
          <artifactId>influxdb-java</artifactId>
      </dependency>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-test</artifactId>
          <scope>test</scope>
      </dependency>

properties配置文件：

server.port=8080
# influxDB
spring.influx.url=http://127.0.0.1:6086
spring.influx.user=admin
spring.influx.password=admin
spring.influx.database=my_sensor1

第二步，增加Influxdb的配置文件；

import java.util.concurrent.TimeUnit;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * influxdb基础配置，也可以直接使用InfluxDB，这么做是为了配置更多的默认参数
 * @author 程就人生
 * @date 2020年7月24日
 * @Description 
 *
 */
@Configuration
public class InfluxdbConfig {
        
    @Value("${spring.influx.url}")
    private String influxDBUrl; 

    @Value("${spring.influx.user}")
    private String userName;    

    @Value("${spring.influx.password}")
    private String password;    

    @Value("${spring.influx.database}")
    private String database;    

    @Bean
    public InfluxDB influxdb(){     
        InfluxDB influxDB = InfluxDBFactory.connect(influxDBUrl, userName, password);
        try {
            
            /** 
             * 异步插入：
             * enableBatch这里第一个是point的个数，第二个是时间，单位毫秒    
             * point的个数和时间是联合使用的，如果满100条或者60 * 1000毫秒   
             * 满足任何一个条件就会发送一次写的请求。
             */
            influxDB.setDatabase(database).enableBatch(100,1000 * 60, TimeUnit.MILLISECONDS);
            
        } catch (Exception e) { 
            e.printStackTrace();
        } finally { 
            //设置默认策略
            influxDB.setRetentionPolicy("sensor_retention");    
        }
        //设置日志输出级别
        influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);  
        return influxDB;
    }
}

第三步，测试代码；

package com.example.controller;

import java.util.ArrayList;
import java.util.List;

import org.influxdb.InfluxDB;
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.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@RestController
public class IndexController {

    @Autowired
    private InfluxDB influxDB;
    
    //measurement
    private final String measurement = "sensor";
    
    @Value("${spring.influx.database}")
    private String database;
    
    /**
     * 批量插入第一种方式
     */
    @GetMapping("/index")
    public void insert(){
        List<String> lines = new ArrayList<String>();       
        Point point = null;     
        for(int i=0;i<50;i++){          
            point = Point.measurement(measurement)
            .tag("deviceId", "sensor" + i)
            .addField("temp", 3)
            .addField("voltage", 145+i)
            .addField("A1", "4i")
            .addField("A2", "4i").build();
            lines.add(point.lineProtocol());
        }
        //写入
        influxDB.write(lines);
    }
    
    /**
     * 批量插入第二种方式
     */
    @GetMapping("/index1")
    public void batchInsert(){      
        BatchPoints batchPoints = BatchPoints
                .database(database)
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
      //遍历sqlserver获取数据
      for(int i=0;i<50;i++){
        //创建单条数据对象——表名
        Point point = Point.measurement(measurement)
          //tag属性——只能存储String类型
                .tag("deviceId", "sensor" + i)
                .addField("temp", 3)
                .addField("voltage", 145+i)
                .addField("A1", "4i")
                .addField("A2", "4i").build();
        //将单条数据存储到集合中
        batchPoints.point(point);
      }
      //批量插入
      influxDB.write(batchPoints); 
    }
    
    /**
     * 获取数据
     */
    @GetMapping("/datas")
    public void datas(@RequestParam Integer page){
        int pageSize = 10;
        // InfluxDB支持分页查询,因此可以设置分页查询条件
        String pageQuery = " LIMIT " + pageSize + " OFFSET " + (page - 1) * pageSize;
        
        String queryCondition = "";  //查询条件暂且为空
        // 此处查询所有内容,如果
        String queryCmd = "SELECT * FROM "
            // 查询指定设备下的日志信息
            // 要指定从 RetentionPolicyName.measurement中查询指定数据,默认的策略可以不加；
            // + 策略name + "." + measurement
            + measurement
            // 添加查询条件(注意查询条件选择tag值,选择field数值会严重拖慢查询速度)
            + queryCondition
            // 查询结果需要按照时间排序
            + " ORDER BY time DESC"
            // 添加分页查询条件
            + pageQuery;
        
        QueryResult queryResult = influxDB.query(new Query(queryCmd, database));        
        log.info("query result => {}", queryResult);
    }
}

第四步，influxdb工具类的编写；

package com.example.config;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.influxdb.InfluxDB;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;
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.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * influxdb操作工具类
 * @author 程就人生
 * @date 2020年7月27日
 * @Description 
 *
 */
@Component
public class InfluxdbUtils {

    @Autowired
    private InfluxDB influxDB;
    
    @Value("${spring.influx.database}")
    private String database;    
    
    /**
     * 新增单条记录,利用java的反射机制进行新增操作
     */
    public void insertOne(Object obj){
        //获取度量
        Class<?> clasz = obj.getClass();
        Measurement measurement = clasz.getAnnotation(Measurement.class);
        //构建
        Point.Builder builder = Point.measurement(measurement.name());
        // 获取对象属性
        Field[] fieldArray = clasz.getDeclaredFields();
        Column column = null;
        for(Field field : fieldArray){
            try {
                column = field.getAnnotation(Column.class);
                //设置属性可操作
                field.setAccessible(true); 
                if(column.tag()){
                    //tag属性只能存储String类型
                    builder.tag(column.name(), field.get(obj).toString());
                }else{
                    //设置field
                    if(field.get(obj) != null){
                        builder.addField(column.name(), field.get(obj).toString());
                    }
                }
            } catch (IllegalArgumentException | IllegalAccessException e) {
                e.printStackTrace();
            }
        }
        influxDB.write(builder.build());
    }
    
    /**
     * 批量新增,方法一
     */
    public void insertBatchByRecords(List<?> records){
        List<String> lines = new ArrayList<String>();   
        records.forEach(record->{
            Class<?> clasz = record.getClass();
            //获取度量
            Measurement measurement = clasz.getAnnotation(Measurement.class);
            //构建
            Point.Builder builder = Point.measurement(measurement.name());
            Field[] fieldArray = clasz.getDeclaredFields();
            Column column = null;
            for(Field field : fieldArray){
                try {
                    column = field.getAnnotation(Column.class);
                    //设置属性可操作
                    field.setAccessible(true); 
                    if(column.tag()){
                        //tag属性只能存储String类型
                        builder.tag(column.name(), field.get(record).toString());
                    }else{
                        //设置field
                        if(field.get(record) != null){
                            builder.addField(column.name(), field.get(record).toString());
                        }
                    }
                } catch (IllegalArgumentException | IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
            lines.add(builder.build().lineProtocol());
        });
        influxDB.write(lines);
    }
    
    /**
     * 批量新增，方法二
     */
    public void insertBatchByPoints(List<?> records){
        BatchPoints batchPoints = BatchPoints.database(database)
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
        records.forEach(record->{
            Class<?> clasz = record.getClass();
            //获取度量
            Measurement measurement = clasz.getAnnotation(Measurement.class);
            //构建
            Point.Builder builder = Point.measurement(measurement.name());
            Field[] fieldArray = clasz.getDeclaredFields();
            Column column = null;
            for(Field field : fieldArray){
                try {
                    column = field.getAnnotation(Column.class);
                    //设置属性可操作
                    field.setAccessible(true); 
                    if(column.tag()){
                        //tag属性只能存储String类型
                        builder.tag(column.name(), field.get(record).toString());
                    }else{
                        //设置field
                        if(field.get(record) != null){
                            builder.addField(column.name(), field.get(record).toString());
                        }
                    }
                } catch (IllegalArgumentException | IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
            batchPoints.point(builder.build());
        });
        influxDB.write(batchPoints);
    }
    
    /**
     * 查询，返回Map集合
     * @param query 完整的查询语句
     * @return
     */
    public List<Object> fetchRecords(String query){
        List<Object> results = new ArrayList<Object>();
        QueryResult queryResult = influxDB.query(new Query(query, database));
        queryResult.getResults().forEach(result->{
            result.getSeries().forEach(serial->{
                List<String> columns = serial.getColumns();
                int fieldSize = columns.size();
                serial.getValues().forEach(value->{     
                    Map<String,Object> obj = new HashMap<String,Object>();
                    for(int i=0;i<fieldSize;i++){   
                        obj.put(columns.get(i), value.get(i));
                    }
                    results.add(obj);
                });
            });
        });
        return results;
    }
    
    /**
     * 查询，返回map集合
     * @param fieldKeys 查询的字段，不可为空；不可为单独的tag
     * @param measurement 度量，不可为空；
     * @param order
     * @param limit
     * @return
     */
    public List<Object> fetchRecords(String fieldKeys, String measurement){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);     
        return this.fetchRecords(query.toString());
    }
    
    /**
     * 查询，返回map集合
     * @param fieldKeys 查询的字段，不可为空；不可为单独的tag
     * @param measurement 度量，不可为空；
     * @param order
     * @param limit
     * @return
     */
    public List<Object> fetchRecords(String fieldKeys, String measurement, String order){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
        query.append(" order by ").append(order);       
        return this.fetchRecords(query.toString());
    }
    
    /**
     * 查询，返回map集合
     * @param fieldKeys 查询的字段，不可为空；不可为单独的tag
     * @param measurement 度量，不可为空；
     * @param order
     * @param limit
     * @return
     */
    public List<Object> fetchRecords(String fieldKeys, String measurement, String order, String limit){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
        query.append(" order by ").append(order);
        query.append(limit);
        return this.fetchRecords(query.toString());
    }
    
    /**
     * 查询，返回对象的list集合
     * @param query
     * @return
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public <T> List<T> fetchResults(String query, Class<?> clasz){
        List results = new ArrayList<>();
        QueryResult queryResult = influxDB.query(new Query(query, database));
        queryResult.getResults().forEach(result->{
            result.getSeries().forEach(serial->{
                List<String> columns = serial.getColumns();
                int fieldSize = columns.size();     
                serial.getValues().forEach(value->{ 
                    Object obj = null;
                    try {
                        obj = clasz.newInstance();
                        for(int i=0;i<fieldSize;i++){   
                            String fieldName = columns.get(i);
                            Field field = clasz.getDeclaredField(fieldName);
                            field.setAccessible(true);
                            Class<?> type = field.getType();
                            if(type == float.class){
                                field.set(obj, Float.valueOf(value.get(i).toString()));
                            }else{
                                field.set(obj, value.get(i));
                            }                           
                        }
                    } catch (NoSuchFieldException | SecurityException | InstantiationException | IllegalAccessException e) {
                        e.printStackTrace();
                    }                   
                    results.add(obj);
                });
            });
        });
        return results;
    }
    
    /**
     * 查询，返回对象的list集合
     * @param fieldKeys
     * @param measurement
     * @param clasz
     * @return
     */
    public <T> List<T> fetchResults(String fieldKeys, String measurement, Class<?> clasz){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);     
        return this.fetchResults(query.toString(), clasz);
    }
    
    /**
     * 查询，返回对象的list集合
     * @param fieldKeys
     * @param measurement
     * @param order
     * @param clasz
     * @return
     */
    public <T> List<T> fetchResults(String fieldKeys, String measurement, String order, Class<?> clasz){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
        query.append(" order by ").append(order);
        return this.fetchResults(query.toString(), clasz);
    }
    
    /**
     * 查询，返回对象的list集合
     * @param fieldKeys
     * @param measurement
     * @param order
     * @param limit
     * @param clasz
     * @return
     */
    public <T> List<T> fetchResults(String fieldKeys, String measurement, String order, String limit, Class<?> clasz){
        StringBuilder query = new StringBuilder();
        query.append("select ").append(fieldKeys).append(" from ").append(measurement);
        query.append(" order by ").append(order);
        query.append(limit);        
        return this.fetchResults(query.toString(), clasz);
    }
}

Java的bean文件不可少，和Mybatis、Hibernate的bean还是有区别的，这里使用了Influxdb-java架包里的注解，针对influxdb的注解；

package com.example.bean;

import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;

import lombok.Data;

//@Builder
@Data
@Measurement(name = "sensor")
public class Sensor {

    @Column(name="deviceId",tag=true)
    private String deviceId;
    
    @Column(name="temp")
    private float temp;
    
    @Column(name="voltage")
    private float voltage;
    
    @Column(name="A1")
    private float A1;
    
    @Column(name="A2")
    private float A2;
    
    @Column(name="time")
    private String time;    
    
}

第五步，测试代码的编写；

package com.example.controller;

import java.util.ArrayList;
import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.example.bean.Sensor;
import com.example.config.InfluxdbUtils;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@RestController
public class IndexController1 {

    @Autowired
    private InfluxdbUtils influxdbUtils;
    
    /**
     * 插入单挑记录
     */
    @GetMapping("/index21")
    public void insert(){
//      Sensor sensor = Sensor.builder().deviceId("0002")
//              .A1(10).A2(10).temp(10L).voltage(10).build();
//      influxdbUtils.insertOne(sensor);
    }
    
    /**
     * 批量插入第一种方式
     */
    @GetMapping("/index22")
    public void batchInsert(){  
        List<Sensor> sensorList = new ArrayList<Sensor>();
        Sensor sensor = null;
//      for(int i=0; i<50; i++){
//          sensor = Sensor.builder().deviceId("000"+i)
//                  .A1(10).A2(10).temp(10).voltage(10).build();
//          sensorList.add(sensor);
//      }
        for(int i=0; i<50; i++){
            sensor = new Sensor();
            sensor.setA1(2);
            sensor.setA2(12);
            sensor.setTemp(9);
            sensor.setVoltage(12);
            sensor.setDeviceId("sensor4545-"+i);
            sensorList.add(sensor);
        }
        influxdbUtils.insertBatchByRecords(sensorList);
    }
    
    /**
     * 批量插入第二种方式
     */
    @GetMapping("/index23")
    public void batchInsert1(){ 
        List<Sensor> sensorList = new ArrayList<Sensor>();
        Sensor sensor = null;
//      for(int i=0; i<50; i++){
//          sensor = Sensor.builder().deviceId("000"+i)
//                  .A1(10).A2(10).temp(10).voltage(10).build();
//          sensorList.add(sensor);
//      }
        for(int i=0; i<50; i++){
            sensor = new Sensor();
            sensor.setA1(2);
            sensor.setA2(12);
            sensor.setTemp(9);
            sensor.setVoltage(12);
            sensor.setDeviceId("sensor4545-"+i);
            sensorList.add(sensor);
        }
        influxdbUtils.insertBatchByPoints(sensorList);
    }
        
    /**
     * 获取数据
     */
    @GetMapping("/datas2")
    public void datas(@RequestParam Integer page){
        int pageSize = 10;
        // InfluxDB支持分页查询,因此可以设置分页查询条件
        String pageQuery = " LIMIT " + pageSize + " OFFSET " + (page - 1) * pageSize;
        
        String queryCondition = "";  //查询条件暂且为空
        // 此处查询所有内容,如果
        String queryCmd = "SELECT * FROM sensor"
            // 查询指定设备下的日志信息
            // 要指定从 RetentionPolicyName.measurement中查询指定数据,默认的策略可以不加；
            // + 策略name + "." + measurement
            // 添加查询条件(注意查询条件选择tag值,选择field数值会严重拖慢查询速度)
            + queryCondition
            // 查询结果需要按照时间排序
            + " ORDER BY time DESC"
            // 添加分页查询条件
            + pageQuery;
        
        List<Object> sensorList = influxdbUtils.fetchRecords(queryCmd);
        log.info("query result => {}", sensorList);
    }
    
    /**
     * 获取数据
     */
    @GetMapping("/datas21")
    public void datas1(@RequestParam Integer page){
        int pageSize = 10;
        // InfluxDB支持分页查询,因此可以设置分页查询条件
        String pageQuery = " LIMIT " + pageSize + " OFFSET " + (page - 1) * pageSize;
        
        String queryCondition = "";  //查询条件暂且为空
        // 此处查询所有内容,如果
        String queryCmd = "SELECT * FROM sensor"
            // 查询指定设备下的日志信息
            // 要指定从 RetentionPolicyName.measurement中查询指定数据,默认的策略可以不加；
            // + 策略name + "." + measurement
            // 添加查询条件(注意查询条件选择tag值,选择field数值会严重拖慢查询速度)
            + queryCondition
            // 查询结果需要按照时间排序
            + " ORDER BY time DESC"
            // 添加分页查询条件
            + pageQuery;
        List<Sensor> sensorList = influxdbUtils.fetchResults(queryCmd, Sensor.class);
        //List<Sensor> sensorList = influxdbUtils.fetchResults("*", "sensor", Sensor.class);
        sensorList.forEach(sensor->{
            log.info("query sensor => {}", sensor);
        });     
    }
}