在之前的文章中可能已经对相关概念有所了解。目前SpringBoot集成InfluxDB 2.x时需要注意InfluxDB 2.x尚未支持SQL查询。对于习惯使用SQL的开发人员来说直接使用Flux语言可能会存在一定的不适应目录一、InfluxDB导入时序数据二、 SpringBoot集成InfluxDB2.x1. maven依赖2. yml配置3. 配置类4. 工具类5. 实体类6. 模板类7. 实现类8. 测试三、调试过程中踩的坑1、Flux语句2、InfluxDB数据转POJO3、Flux常见报错查阅了CSDN上的相关资料发现大多数文档都是基于Flux实现的。考虑到实际开发需求决定重新编写相关的工具类来简化操作流程还是直接操作起来更容易掌握InfluxDB2.x相关文章SpringBoot集成InfluxDB 2.x 以及常用方法封装这篇写的还可以但是有一些不太熟悉的类弃之InfluxDB查询构建组件还是觉得有点不够简洁易懂但有源代码https://gitee.com/lichenpark/influx-query-wrapper可以参考这两篇文章CDSN好多写的什么玩意……一、InfluxDB导入时序数据版本主要查询语言说明InfluxDB 1.xInfluxQL早期版本使用的 SQL-like 语言。InfluxDB 2.xFlux当前使用的版本官方主推Flux查询。InfluxDB 3.xSQL / InfluxQL最新的重构版本不再支持 Flux转向原生 SQLFlux语句data from(bucket: example-bucket) | range(start: -1h) | filter(fn: (r) r._measurement example-measurement and r._field example-field)看上去还是和SQL区别挺大的官方文档https://docs.influxdata.com/influxdb/v2/query-data/flux/#example-data-variable这个语句的含义是从名为example-bucket的存储桶中读取最近 1 小时内测量名称为example-measurement且field字段名为example-field的所有时序数据点from从哪一个桶中查询数据range查询最近一小时的数据filter: 根据字段、标签或任何其他列值查询数据类似 SQL 的查询语言中的SELECT语句和WHERE子句等同于MySQL语句:SELECT _time, _value FROM example-bucket.example-measurement WHERE _time NOW() - INTERVAL 1 HOUR AND _field example-field;个人认为没有必要单独去学习Flux语句在下面实践中就会熟悉Flux语言本文还是使用官方提供的空气传感器示例数据因为数据量每天都有https://docs.influxdata.com/influxdb/v2/reference/sample-data/#air-sensor-sample-data下载后的数据格式从当前表可以看出measurement : airSensorsfieldco、humidity、temperaturetagsensor_id三种方式1、CSVhttps://github.com/influxdata/influxdb2-sample-data/tree/master2、Cliinflux write --bucket echola-bucket --url https://influx-testdata.s3.amazonaws.com/air-sensor-data-annotated.csv3、直接从官方文档下载非最新数据通过CSV下载只能下载当天的数据如果保存每天的运行数据需要创建InfluxDB Task后面再说……还是通过InfluxDB Web UI的Load Data——Source——File Upload——Upload a CSV可参照上一篇文章InfluxDB二——内存原理核心概念以及与MySQL差异通俗解析二、 SpringBoot集成InfluxDB2.x踩了好多坑无语了利用AI生成了一些代码还得调半天还好通了……只展示部分代码有些没验证过就不放上来了一通百通啦要源代码的可以私我目录文件大概是1. maven依赖!-- InfluxDB2 官方客户端 -- dependency groupIdcom.influxdb/groupId artifactIdinfluxdb-client-java/artifactId version5.0.0/version /dependency !-- Hutool 工具类 -- dependency groupIdcn.hutool/groupId artifactIdhutool-all/artifactId version5.8.5/version /dependency dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId version1.8.24/version /dependency2. yml配置相关参数获取请参照前文spring: influxdb: url: http://localhost:8086 token: your token org: your org bucket: your bucket3. 配置类package com.echola.influxdblearning.config; import com.influxdb.client.InfluxDBClient; import com.influxdb.client.InfluxDBClientFactory; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * Author: echola * Date: 2026/4/21 15:58 * Description: */ Data Configuration ConfigurationProperties(prefix influx) public class InfluxDBConfig { /** * 连接地址 */ private String url; /** * 认证token */ private String token; /** * 组织 */ private String org; /** * 数据库 */ private String bucket; /** * 创建 InfluxDB 客户端 Bean */ Bean public InfluxDBClient influxDBClient() { return InfluxDBClientFactory.create(url, token.toCharArray(), org, bucket); } }4. 工具类官方提供的wiki文档https://deepwiki.com/influxdata/influxdb-client-java/官方提供Example :https://github.com/influxdata/influxdb-client-java/blob/9e0ec0be187bdcdab4c03cdb7ded30201e61db6c/client/README.md?plain1#L647Flux语句拼接package com.echola.influxdblearning.utils; import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.date.DatePattern; import cn.hutool.core.date.DateUtil; import cn.hutool.core.util.StrUtil; import cn.hutool.db.sql.Direction; import com.echola.influxdblearning.enums.FluxEnum; import com.influxdb.annotations.Measurement; import java.util.Date; import java.util.List; /** * Author: echola * Date: 2026/4/22 10:16 * Description: Flux语句拼接 */ public class FluxUtil { private final String flux; public String getFlux() { return flux; } Override public String toString() { return FluxUtil{ flux flux }; } private FluxUtil(Builder builder) { flux builder.flux.toString(); } public static class Builder { private final StringBuilder flux; private Class? measurementClass; public Builder() { this.flux new StringBuilder(); } public FluxUtil build() { return new FluxUtil(this); } /** * 设置桶 */ public Builder bucket(String bucket) { flux.append(from(bucket: \).append(bucket).append(\)); return this; } public T Builder measurement(ClassT clazz) { // 直接从注解拿不用任何硬编码 this.measurementClass clazz; Measurement measurement clazz.getAnnotation(Measurement.class); if (measurement null) { throw new IllegalArgumentException(类 clazz.getName() 未添加 Measurement 注解); } String name measurement.name(); flux.append( | filter(fn: (r) r._measurement \).append(name).append(\)); return this; } /** * 时间范围查询 */ public Builder timeRange(Date startTime, Date endTime) { // Hutool 格式化2026-04-22T14:30:00Z String start DateUtil.format(startTime, DatePattern.UTC_PATTERN); String end DateUtil.format(endTime, DatePattern.UTC_PATTERN); if (StrUtil.isNotBlank(start) StrUtil.isNotBlank(end)) { flux.append( | range(start: ).append(start).append(, stop: ).append(end).append()); } else if (StrUtil.isNotBlank(start)) { flux.append( | range(start: ).append(start).append()); } else if (StrUtil.isNotBlank(end)) { flux.append( | range(stop: ).append(end).append()); } return this; } /** * 判断字段是 Tag 还是 Field */ private boolean isTag(String fieldName) { if (measurementClass null) { throw new IllegalStateException(请先调用 setMeasurement(Class) 方法); } // 遍历所有字段根据 Column(name) 匹配而不是 Java 字段名 for (java.lang.reflect.Field field : measurementClass.getDeclaredFields()) { com.influxdb.annotations.Column column field.getAnnotation(com.influxdb.annotations.Column.class); if (column null) { continue; } // 获取注解里的数据库列名 String columnName column.name(); // 如果注解没写 name则用字段名兼容默认 if (columnName.isEmpty()) { columnName field.getName(); } // 匹配传入的 fieldName如 sensor_id if (columnName.equals(fieldName)) { return column.tag(); } } // 遍历完都没找到抛出异常 throw new IllegalArgumentException(类 measurementClass.getName() 中不存在字段: fieldName); } /** * 条件过滤 */ public Builder filter(String key, String operator, String value) { flux.append( | filter(fn: (r) ); if (isTag(key)) { // Tag: r.tagKey value flux.append(r.).append(key).append( ).append(operator).append( \).append(value).append(\); } else { // Field: r._field fieldName and r._value xxx flux.append(r._field \).append(key) .append(\ and r._value ).append(operator).append( ).append(value); } flux.append()); return this; } /** * 行转列 * 标准写法按时间分组将 _field 列展开为多个字段列 */ public Builder pivot() { flux.append( | pivot(rowKey: [\_time\], columnKey: [\_field\], valueColumn: \_value\)); return this; } /** * 限制返回条数 */ public Builder limit(long n) { flux.append( | limit(n: ).append(n).append()); return this; } /** * 分页 */ public Builder page(Integer pageNum, Integer pageSize) { if (pageNum 1) pageNum 1; Integer offset (pageNum - 1) * pageSize; flux.append( | limit(n: ).append(pageSize).append(, offset: ).append(offset).append()); return this; } } }5. 实体类映射空气传感器InfluxDB数据注意此处的Time的类型一定是InstantInfluxDB2.x不支持String|Datepackage com.echola.influxdblearning.entity.po; import com.influxdb.annotations.Column; import com.influxdb.annotations.Measurement; import lombok.Data; import java.time.Instant; /** * Author: echola * Date: 2026/4/21 16:41 * Description: */ Data Measurement(name airSensors) public class AirSensorData { Column(name sensor_id, tag true) private String sensorId; Column(name temperature) private Double temperature; Column(name humidity) private Double humidity; Column(name co) private Double co; Column(timestamp true) private Instant time; }6. 模板类package com.echola.influxdblearning.template; import com.echola.influxdblearning.utils.FluxUtil; import com.influxdb.client.InfluxDBClient; import com.influxdb.client.QueryApi; import com.influxdb.client.WriteApi; import com.influxdb.client.WriteApiBlocking; import com.influxdb.client.domain.WritePrecision; import com.influxdb.client.write.Point; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import java.util.List; /** * Author: echola * Date: 2026/4/21 18:35 * Description: */ Component public class InfluxDBTemplate { Autowired private InfluxDBClient influxDBClient; public T ListT queryList(FluxUtil.Builder builder, ClassT clazz) { String flux builder.build().getFlux(); return query(flux, clazz); } public T ListT query(String flux, ClassT clazz) { return queryApi().query(flux, clazz); } }7. 实现类需求查询指定时间范围内传感器TLM0100的数据按时间倒序再分页返回public ListAirSensorDataVO queryByTimeRange(AirSensorDataDTO dto) { FluxUtil.Builder builder new FluxUtil.Builder().bucket(bucket) .timeRange(dto.getStartTime(), dto.getEndTime()) .measurement(AirSensorData.class) .filter(sensor_id, , dto.getSensorId()) .pivot() .sort(Direction.DESC) .page(dto.getPageSize(), dto.getPageNum()); ListAirSensorData dataList influxDBTemplate.queryList(builder, AirSensorData.class); return ConvertUtil.sourceToTarget(dataList, AirSensorDataVO.class); }对应的FLux语句from(bucket: echola-bucket) | range(start: 2026-04-01T08:00:00Z, stop: 2026-04-23T12:00:00Z) | filter(fn: (r) r._measurement airSensors) | filter(fn: (r) r.sensor_id TLM0100) | pivot(rowKey: [_time], columnKey: [_field], valueColumn: _value) | sort(columns: [_time], desc: true) | limit(n: 10, offset: 0)可以在InfluxDB Web UI测试一下Flux语句数据正常返回8. 测试通过PostMan调接口可以看到数据正确返回啦三、调试过程中踩的坑1、Flux语句关于Flux查询语句的顺序非常重要核心是按顺序串联管道操作各个函数来处理数据数据会从一个函数流向另一个函数就像MySQL也是有顺序的Select……From……Where……基本规则与写法固定开端查询通常以from()定义数据库桶开始。必须限定时间range()函数不可或缺必须指定查询的时间范围否则查询不会执行。管道串联使用|符号将一个函数的输出传给下一个函数像流水线一样处理数据。动态过滤在filter()函数中使用fn: (r) ...的写法来筛选数据行1基本结构Flux语句的顺序基本的结构顺序如下bucket echola-bucket start -1h stop now() from(bucket: bucket) | range(start: start, stop: stop) | filter(fn: (r) r._measurement airSensors) | filter(fn: (r) r.sensor_id TLM001) | filter(fn: (r) r._field temperature or r._field humidity or r._field co) | aggregateWindow(every: 1m, fn: mean) | sort(columns: [_time], desc: true) | limit(n: 10) | yield()Flux语句严格按照以上顺序执行不能混淆顺序日常查询最简固定套路from → range → filter → aggregateWindow → sort → limit → yield虽然Flux是从上往下执行的但是它不像MySQL一样先查询所有数据再进行range和filter。它会做一个关键优化谓词下推具体表现是range()和filter()这类过滤条件会被引擎提前执行引擎直接到存储层只读取满足时间范围和过滤条件的数据Block其余数据不会加载内存所以物理层面range()和filter()是最先且同时起作用的已经进行了一次高效的索引扫描因此将range、filter等尽快放在前面它们能被下推到底层存储引擎执行大幅减少后续处理的数据量如果你先 filter后 range数据范围不会被提前裁剪扫描全表→巨慢结果可能不正确生产环境直接拖垮数据库那现在再一行一行来看吧看完Flux掌握99%哈哈哈哈……①from(bucket: bucket)必须作用指定从哪个桶查询必须写在最前面没有之一②| range(start: start, stop: stop)必须作用限定时间范围必须紧跟 from没有它直接报错unbounded read时间格式支持:now()、-15s、-1h、-7d、2025-01-01T00:00:00Z③| filter(fn: (r) r._measurement airSensors)必须作用指定查询哪张表第一个 filter④| filter(fn: (r) r.sensor_id TLM001)作用按 Tag 过滤⑤| filter(fn: (r) r._field temperature or r._field humidity or r._field co)作用按照Field过滤只查询需要的字段提升速度如果只需要筛选温度在50~60那就只用读取temperature这一列| filter(fn: (r) r._field temperature) | filter(fn: (r) r._value 50 and r._value 60)可以看到Tag是r.tag名称而field是通过r._field和r._value进行过滤不要在filter函数里直接做字符串拼接或数学运算破坏性能优化仅用于过滤⑥| aggregateWindow(every: 1m, fn: mean)作用聚合1 分钟取一个平均值用于数据降采样⑦| sort(columns: [_time], desc: true)作用按时间降序⑧| limit(n: 10)作用只返回 10 条可用于分页。n每页条数offset跳过多少条从 0 开始⑨| yield()作用表示最终结果返回,单个结果可省略yield()多个结果时必须用yield(name: 结果名)为每个输出命名Flux 没有for或if-else但提供了类似三元运算符的if条件表达式如a if true then 1 else 02、InfluxDB数据转POJO①行转列由于InfluxDB的数据是列式存储的不是像MySQL中行式存储详情可见上篇文章:InfluxDB 默认存储是长表直接看真实数据执行Flux语句from(bucket: echola-bucket) | range(start: 2026-04-23T00:00:00Z, stop: 2026-04-24T00:00:00Z) | filter(fn: (r) r._measurement airSensors) | filter(fn: (r) r.sensor_id TLM0100) | sort(columns: [_time], desc: true) | limit(n: 3, offset: 0)可以看到数据格式是数据是可以看到是2026-04-23 10:01:51同一个时间点的3条Field温度、湿度、CO的数据由于不同Field是分开文件存储并不像MySQL行式存储_time _field _value sensor_id -------------------------------------------------------- 2026-04-23 10:01:51 temperature 71.38 TLM0100 2026-04-23 10:01:51 humidity 35.19 TLM0100 2026-04-23 10:01:51 co 61.92 TLM0100再来看看接口的真实效果还是上面那个接口api/air-sensors/query-by-time-range返回返回JSON:[ { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:51 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:41 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:31 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:51 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:41 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:31 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:51 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:41 }, { sensorId: TLM0100, temperature: null, humidity: null, co: null, time: 2026-04-23 10:01:31 } ]是不是发现不对了理论上应该只返回3行数据但却返回了9条数据也缺少了Field字段跟想象的不一样吧Web UI是有Field字段的InfluxDB中筛选后的真实数据是这样的time sensor_id _field _value 2026-04-23 10:01:51 TLM0100 temperature 25.5 2026-04-23 10:01:51 TLM0100 humidity 60.0 2026-04-23 10:01:51 TLM0100 co 0.03 2026-04-23 10:01:41 TLM0100 temperature 25.4 2026-04-23 10:01:41 TLM0100 humidity 60.1 2026-04-23 10:01:41 TLM0100 co 0.03 2026-04-23 10:01:31 TLM0100 temperature 25.3 2026-04-23 10:01:31 TLM0100 humidity 60.2 2026-04-23 10:01:31 TLM0100 co 0.03❓为什么limit(3)返回 9 条这就跟 一篇文章InfluxDB二——内存原理核心概念以及与MySQL差异通俗解析的InfluxDB的存储结构对应上了在物理层面上sensor_idTLM0100是三份完全隔离的物理数据块存储块标识内容简述Shard 1 / TSM FileTLM0100的时间戳列表 co压缩值TLM0100的时间戳列表 humidity压缩值TLM0100的时间戳列表 temperature压缩值那么3个字段3个时间点 3*3 9条数据limit 是限制「行」不是限制「时间点」由于Field存储在不同的Blocklimit依次读取 3 个独立物理文件temperature、humidity、co每个文件取前3条3 3 3 9 条数据❓为什么Field字段全是null因为数据是“长格式”属性名temperature、humidity在_field列里值在_value列里。Java 实体类找的是temperature列当然找不到映射不上就为null理论上应该是——3个时间点的数据3条数据才对_time temperature humidity co ----------------------------------------------------- 2026-04-23 10:01:51 71.38 35.19 61.92 2026-04-23 10:01:50 71.35 35.20 61.90 2026-04-23 10:01:49 71.32 35.18 61.89那如何调整成下面这样的格式呢✅ 解决办法必须加pivot()它对已加载到内存中的原始数据按_time分组对齐将多行结构转换为宽表结构| pivot(rowKey:[_time], columnKey: [_field], valueColumn: _value)含义是按时间分组把字段名变成列把值填入列最终from(bucket: echola-bucket) | range(start: 2026-04-23T00:00:00Z, stop: 2026-04-24T00:00:00Z) | filter(fn: (r) r._measurement airSensors) | filter(fn: (r) r.sensor_id TLM0100) | pivot(rowKey:[_time], columnKey: [_field], valueColumn: _value) | sort(columns: [_time], desc: true) | limit(n: 3, offset: 0)注意pivot的顺序是在filter之后sort|limit之前因为pivot会将符合筛序条件的数据3个文件按照时间合并后再进行排序和limit。pivot 发生在查询引擎的内存计算层 属于数据结构重塑操作不涉及任何物理存储层的修改或合并②时间格式是Instant注意此处的Time的类型一定是InstantInfluxDB2.x不支持String|Date2026-04-01T08:00:00ZJava 对应Instant.now().toString()3、Flux常见报错1.cannot submit unbounded read原因没写 range 或顺序错了解决from 后面紧跟 range2.undefined record原因字段名写错解决检查 Tag/Field 名称