InfluxDB 集成指南

Bucket（存储桶）
 └─ Measurement（测量，类似"表"）
     ├─ Tag（索引列，字符串，用于过滤）
     ├─ Field（数据列，数值/字符串/布尔，存实际值）
     └─ Timestamp（时间戳，主键的一部分）

from(bucket: "demo-bucket")                           // 数据源
  |> range(start: -1h)                                // 时间窗口（必选）
  |> filter(fn: (r) => r._measurement == "device_realtime")       // 测量过滤
  |> filter(fn: (r) => r.device_id == "101")          // tag 过滤
  |> group(columns: ["_field"])                       // 分组
  |> last()                                           // 聚合
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")  // 长转宽

# 1. 安装后访问 http://<host>:8086 完成初始化
# 2. 创建组织（org）和存储桶（bucket），bucket 按数据保留期设置（如 90 天）
# 3. 为应用生成 API Token（建议只授予目标 bucket 的读写权限）

<dependency>
    <groupId>com.influxdb</groupId>
    <artifactId>influxdb-client-java</artifactId>
    <version>6.12.0</version>
</dependency>

# application.yml
influxdb:
  enabled: true                        # 总开关，false 时应用可启动、读写时抛业务异常
  url: http://localhost:8086
  token: ${INFLUXDB_TOKEN:}            # 生产/测试环境用环境变量注入，禁止明文提交
  org: demo-org
  bucket: demo-bucket
  connect-timeout: 10000               # 连接超时（毫秒）
  read-timeout: 10000
  write-timeout: 10000
  gzip: true                           # 官方建议开启，写入可提速数倍
  log-level: NONE                      # NONE/BASIC/HEADERS/BODY
  write-precision: MS                  # 时间戳精度，取可接受的最粗精度
  health-check: true                   # 启动时健康检查
  # ---- 批量写入调优 ----
  write-batch-size: 1000               # 每批数据点数（官方建议 1000~10000）
  write-flush-interval: 1000           # 刷盘间隔（毫秒）
  write-buffer-limit: 10000            # 缓冲区上限，必须大于 batch-size
  write-jitter-interval: 0             # 多实例部署时的随机抖动
  # ---- HTTP 并发调优 ----
  max-requests-per-host: 32            # OkHttp 默认仅 5，大屏轮询会客户端排队
  max-idle-connections: 20
  keep-alive-duration: 5               # 分钟

// config/properties/InfluxDbProperties.java
@Data
@ConfigurationProperties(prefix = "influxdb")
public class InfluxDbProperties {
    private Boolean enabled = true;
    private String url = "http://localhost:8086";
    private String token;
    private String org;
    private String bucket;
    // ... 超时、gzip、批量写、并发等（见 3.3 配置项）
}

// config/InfluxDbConfig.java
@Slf4j
@AutoConfiguration
@EnableConfigurationProperties(InfluxDbProperties.class)
public class InfluxDbConfig {

    @Bean(destroyMethod = "close")
    @ConditionalOnProperty(prefix = "influxdb", name = "enabled",
                           havingValue = "true", matchIfMissing = true)
    public InfluxDBClient influxDBClient(InfluxDbProperties properties) {
        // ① 必要参数快速失败：url/token/org/bucket 任一为空直接抛异常
        // ② 构建 InfluxDBClientOptions（url、token 认证、org、bucket、精度）
        // ③ OkHttpClient 定制：超时 + Dispatcher 并发 + 连接池
        // ④ 工厂创建 client，按需开启 gzip
        // ⑤ 启动健康检查（health() 可拿到 status/version）
    }

    @Bean(destroyMethod = "close")
    @ConditionalOnBean(InfluxDBClient.class)
    public WriteApi influxDBWriteApi(InfluxDBClient client, InfluxDbProperties properties) {
        // ① WriteOptions 配置化（batchSize/flushInterval/bufferLimit/jitter）
        // ② 注册三个写失败事件监听器（见 4.2）
    }

    @Bean
    @ConditionalOnBean(InfluxDBClient.class)
    public QueryApi influxDBQueryApi(InfluxDBClient client) {
        return client.getQueryApi();
    }

    /**
     * Helper 无条件注册（ObjectProvider 软依赖注入三个 API），
     * enabled=false 时进入降级状态而非启动崩溃
     */
    @Bean
    public InfluxDbHelper influxDbHelper(...) { ... }
}

// util/InfluxDbHelper.java
public class InfluxDbHelper {

    // ---- 写入（异步批量语义）----
    public void writePoint(Point point);
    public void writePoints(List<Point> points);

    // ---- 查询 ----
    public List<FluxTable> query(String flux);
    public List<FluxTable> query(String flux, Map<String, Object> params);  // 参数化

    // ---- 删除（predicate 仅支持 _measurement 和 tag）----
    public void delete(String bucket, Instant start, Instant stop, String predicate);

    // ---- 值提取（容错脏数据，返回 null 而非抛异常）----
    public static BigDecimal getBigDecimal(Map<String, Object> values, String key);
    public static Long getLong(Map<String, Object> values, String key);
    public static String getString(Map<String, Object> values, String key);
    public static Instant getInstant(Map<String, Object> values, String key);

    // ---- Flux 字符串转义 ----
    public static String escapeFluxString(String value);
}

// 构建 Point：维度用 tag，数值用 field，null 的 field 直接跳过
Point point = Point.measurement("device_realtime")
    .addTag("tenant_id", data.getTenantId())
    .addTag("site_id", String.valueOf(data.getSiteId()))
    .addTag("device_id", String.valueOf(data.getDeviceId()))
    .time(data.getCollectionTime().toInstant(), WritePrecision.MS)
    .addField("voltage", data.getVoltage())
    .addField("temperature", data.getTemperature())
    .build();

influxDbHelper.writePoint(point);   // 异步批量，调用即返回

from(bucket: params.bucket)
  |> range(start: -168h)                              // 只扫合理窗口（如 7 天），设备离线超窗口即视为离线
  |> filter(fn: (r) => r._measurement == params.measurement)
  |> filter(fn: (r) => r.device_id == params.deviceId)
  |> group(columns: ["_field"])
  |> last()                                           // 每字段只留最新 1 条
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")

from(bucket: params.bucket)
  |> range(start: params.start, stop: params.end)
  |> filter(fn: (r) => r._measurement == params.measurement)
  |> filter(fn: (r) => r.device_id == params.deviceId)
  |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> sort(columns: ["_time"])

|> filter(fn: (r) => contains(value: r.device_id, set: ["101", "102", "103"]))

List<FluxTable> tables = influxDbHelper.query(flux, params);
for (FluxTable table : tables) {
    for (FluxRecord record : table.getRecords()) {
        Map<String, Object> values = record.getValues();
        vo.setVoltage(InfluxDbHelper.getBigDecimal(values, "voltage"));
        vo.setCollectionTime(InfluxDbHelper.getInstant(values, "_time"));
    }
}

公共基础模块（如 common-influxdb）
 ├─ config/InfluxDbConfig.java          # 自动配置：client/writeApi/queryApi/helper 四个 Bean
 ├─ config/properties/InfluxDbProperties.java
 └─ util/InfluxDbHelper.java            # 统一读写入口 + 值转换 + 转义

业务模块
 ├─ repository/DeviceRealtimeInfluxRepository.java   # Flux 查询构建 + VO 转换
 └─ service/impl/...                    # 写入（buildPoint）+ 业务编排

// 必须注册的三个监听器（官方 README: Monitoring & Alerting）
writeApi.listenEvents(WriteErrorEvent.class, event ->
    log.error("异步写入失败（4xx 不可重试，批次已丢弃）", event.getThrowable()));
writeApi.listenEvents(WriteRetriableErrorEvent.class, event ->
    log.error("写入重试耗尽（5xx，批次已丢弃）", event.getThrowable()));
writeApi.listenEvents(BackpressurePressEvent.class, event ->
    log.warn("缓冲区背压，新数据可能被丢弃"));

influx delete --bucket demo-bucket --org demo-org \
  --start 1970-01-01T00:00:00Z --stop 2099-01-01T00:00:00Z \
  --predicate '_measurement="device_realtime"'

|> range(start: -8760h)              # 扫一年
|> pivot(...)                        # 全量 pivot 成宽表
|> sort(columns: ["_time"], desc: true)
|> limit(n: 1)

Dispatcher dispatcher = new Dispatcher();
dispatcher.setMaxRequestsPerHost(32);
dispatcher.setMaxRequests(Math.max(64, 32));   // 全局上限需不小于单主机上限
okHttpBuilder.dispatcher(dispatcher);