说说设备监控数据怎么存、怎么查？Spring Boot + TimescaleDB 完整实战（学习笔记）

今天翻到一篇不错的技术分享，看完之后自己也琢磨了一下，把思路梳理记录下来。

示例环境：JDK 17、Spring Boot 3.x、PostgreSQL 17、TimescaleDB。本文重点是把设备指标的写入、按时间查询和聚合查询串成一条可运行的链路。

写在前面

设备每隔几秒上报一次温度、湿度、压力或电量，数据量不大时，直接存普通关系表当然可以。等设备数量和保留时间增加后，大家通常会开始问：历史数据怎么按时间查？报表每分钟聚合一次会不会越来越慢？旧数据如何清理？

这类数据有一个共同特点：每条记录都带着时间戳，写入频繁，查询通常带时间范围，还经常需要按分钟、小时做聚合。本文用 TimescaleDB 配合 Spring Boot 做一个小型设备监控服务：设备批量上报指标，服务端写入 hypertable，再按设备和时间范围查询，并演示五分钟聚合。

先把边界说清楚：时序数据库不是“有时间字段就务必上”。数据规模较小、查询简单时，MySQL 或 PostgreSQL 普通表也能胜任。是否采用时序数据库，要结合写入量、保留策略、聚合频率和运维能力来决定。

一、先看整体链路

设备 / 模拟数据
      │ POST /api/measurements 批量上报
      ▼
Spring Boot Controller
      │ 参数校验与业务边界
      ▼
JdbcTemplate 批量写入
      │ INSERT ... ON CONFLICT
      ▼
TimescaleDB hypertable
      │ device_id + time 范围过滤
      ├── 原始数据查询
      └── time_bucket 五分钟聚合

TimescaleDB 基于 PostgreSQL。普通表可以按时间列转换成 hypertable；应用侧仍使用 PostgreSQL JDBC 驱动和 SQL，不需要引入专有查询语言。hypertable 的分块和时间范围剪枝由数据库负责，业务接口仍可保持常用的 REST + JDBC 形态。

二、启动 TimescaleDB

准备 docker-compose.yml：

services:
  timescaledb:
    image: timescale/timescaledb:latest-pg17
    container_name: telemetry-db
    environment:
      POSTGRES_DB: telemetry
      POSTGRES_USER: app
      POSTGRES_PASSWORD: change-me-local
    ports:
      - "5432:5432"
    volumes:
      - timescale_data:/var/lib/postgresql/data
      - ./db/init.sql:/docker-entrypoint-initdb.d/001-init.sql:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app -d telemetry"]
      interval: 5s
      timeout: 3s
      retries: 20

volumes:
  timescale_data:

本地启动：

docker compose up -d
docker compose ps

latest-pg17 方便本地快速体验。生产环境应固定经过验证的镜像版本或 digest，并通过密钥管理系统注入数据库密码，不要把示例密码直接用于线上。

新建 db/init.sql：

CREATE EXTENSION IF NOT EXISTS timescaledb;

CREATE TABLE IF NOT EXISTS measurement (
    time        TIMESTAMPTZ      NOT NULL,
    device_id   TEXT             NOT NULL,
    temperature DOUBLE PRECISION,
    humidity    DOUBLE PRECISION,
    PRIMARY KEY (device_id, time)
);

SELECT create_hypertable(
    'measurement',
    by_range('time'),
    if_not_exists => TRUE
);

这里把 time 设为分区时间列，把 device_id 放进主键。TimescaleDB 要求 hypertable 上的唯一约束包含所有分区列，于是主键采用 (device_id, time)。同一设备同一时间点重复上报时，我们可以用 UPSERT 覆盖原值，避免生成重复记录。

初始化脚本只会在数据库数据目录首次创建时执行。如果容器之前已经启动过，挂载新脚本不会自动补跑；可以通过 docker exec -it telemetry-db psql -U app -d telemetry 进入数据库，再执行脚本内容。

三、创建 Spring Boot 项目

Maven 依赖示例：


    org.springframework.boot
    spring-boot-starter-parent
    3.4.5

    17

        org.springframework.boot
        spring-boot-starter-web

        org.springframework.boot
        spring-boot-starter-jdbc

        org.postgresql
        postgresql
        runtime

这里选择 JdbcTemplate，因为示例重点是时序 SQL 本身。Spring Boot 会根据 JDBC URL 配置 DataSource；当 JDBC starter 和 HikariCP 在依赖中可用时，也会自动配置连接池。

src/main/resources/application.yml：

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/telemetry
    username: app
    password: ${DB_PASSWORD:change-me-local}
    hikari:
      maximum-pool-size: 10
      connection-timeout: 3000

server:
  port: 8080

本地运行时，默认密码与 Compose 示例一致。部署时用环境变量 DB_PASSWORD 注入密码即可。连接池大小需要按数据库连接上限、应用实例数和真实并发量一起评估，不能只因为调大数字就认为吞吐会变好。

四、定义上报数据

每条记录包含设备 ID、采样时间、温度和湿度。时间使用 Instant，接口统一接受带时区的 ISO-8601 时间，例如 2026-10-16T08:30:00Z。

package com.example.telemetry;

import java.time.Instant;

public record MeasurementRequest(
        Instant time,
        String deviceId,
        Double temperature,
        Double humidity
) {}

实际项目还应增加字段校验，例如设备 ID 非空、采样时间不能超出允许窗口、数值范围符合传感器量程。这里先聚焦存储与查询路径。

五、用批量 UPSERT 写入数据

设备上报常常是一批点位。逐条打开事务、逐条执行 INSERT 会增加网络往返和 SQL 执行开销；JdbcTemplate.batchUpdate 可以把一批参数交给 JDBC 批量执行。

package com.example.telemetry;

import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.List;

@Repository
public class MeasurementRepository {
    private static final String UPSERT_SQL = """
        INSERT INTO measurement(time, device_id, temperature, humidity)
        VALUES (?, ?, ?, ?)
        ON CONFLICT (device_id, time)
        DO UPDATE SET
            temperature = EXCLUDED.temperature,
            humidity = EXCLUDED.humidity
        """;

    private final JdbcTemplate jdbcTemplate;

    public MeasurementRepository(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    public int saveBatch(List points) {
        int[] counts = jdbcTemplate.batchUpdate(
                UPSERT_SQL,
                points,
                500,
                (PreparedStatement ps, MeasurementRequest point) -> bind(ps, point)
        );
        return counts.length;
    }

    private void bind(PreparedStatement ps, MeasurementRequest point)
            throws SQLException {
        ps.setTimestamp(1, Timestamp.from(point.time()));
        ps.setString(2, point.deviceId());
        setNullableDouble(ps, 3, point.temperature());
        setNullableDouble(ps, 4, point.humidity());
    }

    private void setNullableDouble(PreparedStatement ps, int index, Double value)
            throws SQLException {
        if (value == null) {
            ps.setNull(index, Types.DOUBLE);
        } else {
            ps.setDouble(index, value);
        }
    }
}

500 是每批参数的分组大小，不是性能承诺。设备上报频率、网络延迟、单条数据大小和数据库配置都会影响合适的批次值，建议压测后再调整。使用 ON CONFLICT 让同一设备、同一时刻的重试变成幂等更新；如果业务要求保留每次上报记录，则应设计独立事件 ID，而不是覆盖旧值。

六、按设备和时间范围查询

先做最常用的原始点查询。查询条件固定为设备 ID 和时间范围，结果按时间升序返回：

public List findRaw(
        String deviceId, Instant from, Instant to) {
    String sql = """
        SELECT time, device_id, temperature, humidity
        FROM measurement
        WHERE device_id = ?
          AND time >= ?
          AND time < ?
        ORDER BY time
        """;

    return jdbcTemplate.query(sql, (rs, rowNum) -> new MeasurementRequest(
            rs.getTimestamp("time").toInstant(),
            rs.getString("device_id"),
            (Double) rs.getObject("temperature"),
            (Double) rs.getObject("humidity")
    ), deviceId, Timestamp.from(from), Timestamp.from(to));
}

时间范围使用左闭右开 [from, to)，相邻查询窗口不会重复包含边界点。不要用 WHERE DATE(time) = ? 这类对时间列套函数的写法做大表过滤；直接对时间列设置上下界，更利于数据库按分区范围裁剪数据。

七、五分钟聚合：从原始点到趋势曲线

监控页面通常不需要把每个采样点都返回给浏览器。像是设备每秒上报一次，查询一天就可能产生 86400 个点；可以用 time_bucket 把数据按五分钟分桶，返回平均值和最大值：

SELECT
    time_bucket(INTERVAL '5 minutes', time) AS bucket,
    device_id,
    AVG(temperature) AS avg_temperature,
    MAX(temperature) AS max_temperature
FROM measurement
WHERE device_id = 'sensor-001'
  AND time >= TIMESTAMPTZ '2026-10-16 00:00:00+00'
  AND time <  TIMESTAMPTZ '2026-10-17 00:00:00+00'
GROUP BY bucket, device_id
ORDER BY bucket;

在 Repository 中封装聚合结果：

package com.example.telemetry;

import java.time.Instant;

public record BucketPoint(
        Instant bucket,
        String deviceId,
        Double averageTemperature,
        Double maxTemperature
) {}

public List findFiveMinuteBuckets(
        String deviceId, Instant from, Instant to) {
    String sql = """
        SELECT time_bucket(INTERVAL '5 minutes', time) AS bucket,
               device_id,
               AVG(temperature) AS avg_temperature,
               MAX(temperature) AS max_temperature
        FROM measurement
        WHERE device_id = ?
          AND time >= ?
          AND time < ?
        GROUP BY bucket, device_id
        ORDER BY bucket
        """;

    return jdbcTemplate.query(sql, (rs, rowNum) -> new BucketPoint(
            rs.getTimestamp("bucket").toInstant(),
            rs.getString("device_id"),
            (Double) rs.getObject("avg_temperature"),
            (Double) rs.getObject("max_temperature")
    ), deviceId, Timestamp.from(from), Timestamp.from(to));
}

如果趋势报表被频繁查询、原始数据又很大，可以进一步评估 TimescaleDB continuous aggregate（连续聚合）。它把分桶聚合结果预先物化，并可配置刷新策略；是否值得启用，要结合数据延迟要求、修正历史数据的频率和维护成本判断。

八、把 Repository 暴露成 REST 接口

package com.example.telemetry;

import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.web.bind.annotation.*;

import java.time.Instant;
import java.util.List;
import java.util.Map;

@RestController
@RequestMapping("/api/measurements")
public class MeasurementController {
    private final MeasurementRepository repository;

    public MeasurementController(MeasurementRepository repository) {
        this.repository = repository;
    }

    @PostMapping
    public Map ingest(
            @RequestBody List points) {
        return Map.of("accepted", repository.saveBatch(points));
    }

    @GetMapping("/five-minute")
    public List queryFiveMinute(
            @RequestParam String deviceId,
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
            Instant from,
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE_TIME)
            Instant to) {
        if (!from.isBefore(to)) {
            throw new IllegalArgumentException("from 必须早于 to");
        }
        return repository.findFiveMinuteBuckets(deviceId, from, to);
    }
}

本地启动应用后，批量写入示例：

curl -X POST http://localhost:8080/api/measurements \
  -H 'Content-Type: application/json' \
  -d '[
    {"time":"2026-10-16T08:00:00Z","deviceId":"sensor-001","temperature":22.6,"humidity":46.2},
    {"time":"2026-10-16T08:00:01Z","deviceId":"sensor-001","temperature":22.8,"humidity":46.0}
  ]'

查询五分钟聚合：

GET http://localhost:8080/api/measurements/five-minute?deviceId=sensor-001&from=2026-10-16T08:00:00Z&to=2026-10-16T09:00:00Z

示例响应结构：

[
  {
    "bucket": "2026-10-16T08:00:00Z",
    "deviceId": "sensor-001",
    "averageTemperature": 22.7,
    "maxTemperature": 22.8
  }
]

九、上线前要补齐的几个工程细节

1. 接口校验与认证： 限制每次批量点数，校验时间范围和指标量程，并验证设备身份，避免任意客户端写入。
2. 写入削峰： 入口流量超过数据库可承载能力时，把接入层和数据库写入解耦，使用消息队列缓冲；不要在 Controller 里无界地创建后台线程。
3. 保留与降采样： 原始数据保存多久、聚合数据保存多久，要按查询需求和合规要求制定；删除历史数据前先确认备份和保留策略。
4. 连接池与批量大小： 结合数据库最大连接数、应用实例数和压测结果调参，避免实例扩容后把数据库连接耗尽。
5. 查询边界： 强制限制最大查询时间跨度和返回点数；趋势图优先返回聚合数据，避免一次把数百万行传给前端。
6. 时间与时区： 数据库存 TIMESTAMPTZ，接口传 ISO-8601 时间，服务端统一用 Instant，不要在多层代码里混用本地时区时间。

十、什么时候选时序数据库？

场景可以先考虑设备少、数据量小、只做简单查询现有关系数据库普通表指标持续写入、查询经常限定时间窗口TimescaleDB 等时序扩展/数据库需要成熟 SQL、事务与既有 PostgreSQL 生态TimescaleDB 这类 PostgreSQL 扩展路线写入吞吐、压缩、保留策略或集群能力有特殊要求根据真实负载对候选产品做压测与选型

不要只看某个数据库宣传的峰值数字。用真实设备数、采样周期、字段数、保留时长和查询比例构造压测数据，再观察写入延迟、查询延迟、磁盘增长和故障恢复时间，结论才和自己的业务有关。

总结

这次实战把一条最小设备监控链路跑通了：Spring Boot 接收批量指标，JdbcTemplate 批量 UPSERT 到 TimescaleDB hypertable，再通过设备 ID、时间范围和 time_bucket 查询趋势数据。

关键点可以归纳为：

用时间列作为 hypertable 的分区维度；唯一约束要包含分区列。
用批量写入减少请求往返，用 UPSERT 处理上报重试。
查询采用 [from, to) 时间范围，避免对过滤列套函数。
趋势图返回聚合数据；热点聚合再评估连续聚合。
先按真实负载判断是否需要时序数据库，不为“技术选型”而选型。

时序数据库真正解决的不是“多一个地方存时间字段”，而是让时间范围写入、保留和聚合查询有一套更贴合数据形态的组织方法。先把链路和访问模式设计清楚，再做容量和性能验证，才是可落地的选型过程。

参考资料

这篇笔记就先到这里，后面用到新的思路或者发现有问题再补充。

小丸子博客