InfluxDB ORM API封装

Posted on by hackdl

InfluxDB 是一个开源的分布式时序、时间和指标数据库，用于存储和检索数据。它是一个自定义的高性能存储引擎，具有时间戳索引和SQL类似的查询语言。
influxdb-client-java 是官方提供的 Java 客户端，但是它的 API 使用起来不是很方便，所以我对其进行了二次封装，使其使用起来更加方便。详细源码

依赖



    com.influxdb
    influxdb-client-java
    6.3.0

配置

使用 InfluxDB 之前，需要先在 InfluxDB 中创建一个 bucket，然后创建一个 token，用于访问该 bucket。

spring:
  influx:
    url: http://localhost:8086
    org: xxx-org
    token: xxx
    bucket: xxx-bucket

注入 InfluxDBClient

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class InfluxDbConfig {

    @Value("${spring.influx.url:''}")
    private String url;

    @Value("${spring.influx.org:''}")
    private String org;

    @Value("${spring.influx.token:''}")
    private String token;

    @Value("${spring.influx.bucket:''}")
    private String bucket;

    @Bean
    public InfluxDBClient influxDB(){
        InfluxDBClient client = InfluxDBClientFactory.create(url, token.toCharArray(), org, bucket);
        return client;
    }
}

注解

为了区分不同实例的相同指标，如cpu0和cpu1等相同的指标，需要在 Measureme服务器托管网nt 上添加前、后缀，如：cpu_measure_0、cpu_measure_1，从而实现区分。
为了自动化实现这个功能，需要在 Measurement 上添加前、后缀注解，如：

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * measurement 前缀
 * 与 @Measurement 拼接 动态的 measurement
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.FIELD})
public @interface MeasurementPrefix {
}
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * measurement 后缀
 * 与 @Measurement 拼接 动态的 measurement
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.FIELD})
public @interface MeasurementSuffix {
}

ORM 工具类

完成POJO与InfluxDB的转换，包括前后缀的拼接，以及常用的 Query、Write 等操作。

import com.influxdb.annotations.Column;
import com.influxdb.annotations.Measurement;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import com.sme.smebackend.influx.MeasurementPrefix;
import com.sme.smebackend.influx.MeasurementSuffix;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Slf4j
public class InfluxDBUtil {
    public static String measurementName(Object o) {
        Class clazz = o.getClass();

        // 获取 @Measurement
        Measurement meaAnno = (Measurement) clazz.getAnnotation(Measurement.class);
        if (null == meaAnno) {
            return null;
        }

        String measurement = meaAnno.name();

        // 获取 @MeasurementPrefix 字段的值
        String prefix = getFieldValueByAnno(o, MeasurementPrefix.class);

        // 获取 @MeasurementSuffix 字段的值
        String suffix = getFieldValueByAnno(o, MeasurementSuffix.class);

        if (null != prefix) {
            measurement = prefix + "_" + measurement;
        }

        if (null != suffix) {
            measurement = measurement + "_" + suffix;
        }

        return measurement;
    }

    @Nullable
    private static String getFieldValueByAnno(Object o, Class annoClazz) {
        Class clazz = o.getClass();
        String val = null;

        for (Field field : clazz.getDeclaredFields()) {
            Annotation anno = field.getAnnotation(annoClazz);
            if (null != anno) {
                try {
              服务器托管网      field.setAccessible(true);
                    val = (String) field.get(o);
                } catch (IllegalAccessException e) {
                    log.error("failed to get value of field: {}", field.getName(), e);
                }
            }
        }

        return val;
    }

    /**
     * 将 @Measurement 注解的对象转换为 Influxdb Point 对象
     */
    public static Point measurement2Point(Object o) {
        String measurement = measurementName(o);
        if (null == measurement) {
            return null;
        }

        Class clazz = o.getClass();

        // 遍历所有的 @Column 字段
        Point point = Point.measurement(measurement);
        for (Field field : clazz.getDeclaredFields()) {
            Column colAnno = field.getAnnotation(Column.class);
            MeasurementPrefix prefixAnno = field.getAnnotation(MeasurementPrefix.class);
            MeasurementSuffix suffixAnno = field.getAnnotation(MeasurementSuffix.class);
            if (null == colAnno || null != suffixAnno || null != prefixAnno) {
                continue;
            }

            try {
                field.setAccessible(true);

                String name = colAnno.name();
                boolean tag = colAnno.tag();
                if (tag) {
                    point.addTag(field.getName(), (String) field.get(o));
                } else {
                    if (Objects.equals("time", name)) {
                        point.time((Instant) field.get(o), WritePrecision.MS);
                    } else {
                        point.addField(field.getName(), (long) field.get(o));
                    }
                }
            } catch (IllegalAccessException e) {
                log.error("failed to get value of field: {}", field.getName(), e);
            }
        }

        return point;
    }

    /**
     * 写入
     */
    public static void write(InfluxDBClient influxDB, Object measurement) {
        WriteApiBlocking w = influxDB.getWriteApiBlocking();
        Point point = measurement2Point(measurement);
        if (null == point) {
            return;
        }

        w.writePoint(point);
    }

    /**
     * 查询某个 measurement 的某个字段
     */
    public static  List query1d(InfluxDBClient influxDB,
                                      String bucket,
                                      String measurement, Class clazz) {
        return query1d(influxDB, bucket, measurement, null, clazz);
    }

    public static  List query1d(InfluxDBClient influxDB,
                                      String bucket,
                                      String measurement, List fields, Class clazz) {
        String query = baseQ(bucket, measurement, fields, "-1d");
        return doQuery(influxDB, clazz, query);
    }

    private static  List doQuery(InfluxDBClient influxDB, Class clazz, String query) {
        List ts = influxDB.getQueryApi().query(query);
        List res = InfluxDBUtil.fluxTable2Pojo(ts, clazz);
        return res;
    }

    @NotNull
    private static String baseQ(String bucket, String measurement, List fields, String start) {
        String query = "from(bucket: "" + bucket + "")" +
                " |> range(start: " + start + ")" +
                " |> filter(fn: (r) => r._measurement == "" + measurement + "")";

        if (!J.empty(fields)) {
            query += " |> filter(fn: (r) => ";

            for (int i = 0; i  fields,
                               String start, String every) {
        String query = baseQ(bucket, measurement, fields, start);
        query += " |> aggregateWindow(every: " + every + ", fn: mean, createEmpty: true)";
        return query;
    }

    public static  List queryAggr(InfluxDBClient influxDB,
                                        String bucket, String measurement, Class clazz,
                                        String start, String every) {
        String query = aggrQ(bucket, measurement, null, start, every);
        return doQuery(influxDB, clazz, query);
    }

    /**
     * 通用聚合查询
     */
    public static  List queryAggr(InfluxDBClient influxDB,
                                        String bucket, String measurement, List fields, Class clazz,
                                        String start, String every) {
        String query = aggrQ(bucket, measurement, fields, start, every);
        return doQuery(influxDB, clazz, query);
    }

    /**
     * 查询某个 measurement 的某个字段， 整 5 min 的数据，如 10:20， 10:25， 10:30
     */
    public static  List queryBy5Min(InfluxDBClient influxDB,
                                          String bucket,
                                          String measurement, Class clazz) {
        return queryBy5Min(influxDB, bucket, measurement, null, clazz);
    }

    public static  List queryBy5Min(InfluxDBClient influxDB,
                                          String bucket,
                                          String measurement, List fields, Class clazz) {
        return queryAggr(influxDB, bucket, measurement, fields, clazz, "-1d", "5m");
    }

    public static  List queryBy1Min(InfluxDBClient influxDB,
                                          String bucket,
                                          String measurement, Class clazz) {
        return queryBy1Min(influxDB, bucket, measurement, null, clazz);
    }

    public static  List queryBy1Min(InfluxDBClient influxDB,
                                          String bucket,
                                          String measurement, List fields, Class clazz) {
        return queryAggr(influxDB, bucket, measurement, fields, clazz, "-1d", "1m");
    }

    /**
     * 查询某个 measurement 的某个字段， 整点数据，如 10:00， 11:00， 12:00
     */
    public static  List queryByHour(InfluxDBClient influxDB,
                                          String bucket,
                                          String measurement, Class clazz) {
        return queryByHour(influxDB, bucket, measurement, null, clazz);
    }

    public static  List queryByHour(InfluxDBClient influxDB,
                                          String bucket,
                                          String measurement, List fields, Class clazz) {
        return queryAggr(influxDB, bucket, measurement, fields, clazz, "-7d", "1h");
    }

    /**
     * 查询某个 measurement 的某个字段， 整天数据，如 2021-01-01 00:00:00， 2021-01-02 00:00:00
     */
    public static  List queryByDay(InfluxDBClient influxDB,
                                         String bucket,
                                         String measurement, Class clazz) {
        return queryByDay(influxDB, bucket, measurement, null, clazz);
    }

    public static  List queryByDay(InfluxDBClient influxDB,
                                         String bucket,
                                         String measurement, List fields, Class clazz) {
        return queryAggr(influxDB, bucket, measurement, fields, clazz, "-1M", "1d");
    }

    /**
     * 删除一个月前的数据
     */
    public static void deleteBeforeOneMonth(InfluxDBClient influxDB, String bucket, String org, String measurement) {
        OffsetDateTime now = OffsetDateTime.now();

        influxDB.getDeleteApi().delete(
                now.minusYears(2), now.minusMonths(1),
                String.format("_measurement="%s"", measurement),
                bucket, org
        );
    }

    /**
     * 删除所有数据
     */
    public static void deleteAll(InfluxDBClient influxDB, String bucket, String org, String measurement) {
        OffsetDateTime now = OffsetDateTime.now();

        influxDB.getDeleteApi().delete(
                now.minusYears(100), now,
                String.format("_measurement="%s"", measurement),
                bucket, org
        );
    }

    /**
     * 将 FluxTable 转化为 pojo list
     */
    @SneakyThrows
    public static  List fluxTable2Pojo(List tables, Class clazz) {
        List ts = J.list();

        for (FluxTable tab : tables) {
            List recs = tab.getRecords();
            for (int i = 0; i  ts.size() - 1) {
                    t = clazz.newInstance();
                    ts.add(t);
                } else {
                    t = ts.get(i);
                }

                record2Pojo(r, t);
            }
        }

        return ts;
    }

    /**
     * 将 {@link FluxRecord } 转化为 pojo 对象
     */
    public static  T record2Pojo(FluxRecord record, T t) {
        try {
            Class clazz = t.getClass();

            Map recordValues = record.getValues();

            String recordField = record.getField();
            Object value = record.getValue();

            Field[] fileds = clazz.getDeclaredFields();
            for (Field field : fileds) {
                Column colAnno = field.getAnnotation(Column.class);

                MeasurementPrefix prefixAnno = field.getAnnotation(MeasurementPrefix.class);
                MeasurementSuffix suffixAnno = field.getAnnotation(MeasurementSuffix.class);
                if (null == colAnno || null != suffixAnno || null != prefixAnno) {
                    continue;
                }

                field.setAccessible(true);

                String colName = colAnno.name();

                if (Objects.equals("time", colName)) {
                    field.set(t, record.getTime());
                } else if (Objects.equals(colName, recordField)) {
                    if (null == value) {
                        continue;
                    }
                    String type = field.getType().getName();

                    if (colName.equals("writeSectTotal")) {
                        System.out.println("writeSectTotal");
                    }

                    if ((type.equals("long") || type.equals("java.lang.Long"))
                            && value instanceof Double) {
                        field.set(t, ((Double) value).longValue());
                    } else if ((type.equals("int") || type.equals("java.lang.Integer"))
                            && value instanceof Double) {
                        field.set(t, ((Double) value).intValue());
                    } else {
                        field.set(t, value);
                    }

                } else {
                    Object v = recordValues.get(field.getName());
                    if (null != v) {
                        field.set(t, v);
                    }
                }
            }

            return t;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 将 {@link FluxRecord } 转化为 pojo 对象
     */
    public static  T record2Pojo(FluxRecord record, Class clazz) {
        try {
            T t = clazz.newInstance();

            Map recordValues = record.getValues();

            String recordField = record.getField();
            Object value = record.getValue();

            for (Field field : clazz.getDeclaredFields()) {
                Column colAnno = field.getAnnotation(Column.class);

                MeasurementPrefix prefixAnno = field.getAnnotation(MeasurementPrefix.class);
                MeasurementSuffix suffixAnno = field.getAnnotation(MeasurementSuffix.class);
                if (null == colAnno || null != suffixAnno || null != prefixAnno) {
                    continue;
                }

                field.setAccessible(true);

                String colName = colAnno.name();

                if (Objects.equals("time", colName)) {
                    field.set(t, record.getTime());
                } else if (Objects.equals(colName, recordField)) {
                    field.set(t, value);
                } else {
                    Object v = recordValues.get(field.getName());
                    if (null != v) {
                        field.set(t, v);
                    }
                }
            }

            return t;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }
}

以上的工具是我在使用 InfluxDB 时，总结出来的一些常用的方法，可以直接拿来使用，大大简化了 InfluxDB 的使用。

服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net
机房租用，北京机房租用，IDC机房托管， http://www.fwqtg.net

相关推荐: Springboot简单功能示例-6 使用加密数据源并配置日志

springboot-sample 介绍 springboot简单示例 跳转到发行版 查看发行版说明 软件架构（当前发行版使用） springboot hutool-all 非常好的常用java工具库 官网 maven bcprov-jdk18on 一些加密算…

服务器托管，北京服务器托管，服务器租用，机房机柜带宽租用

服务器托管

咨询：董先生

电话13051898268 QQ/微信93663045！

上一篇:
下一篇: