档案放 PostgreSQL时序放 TDengine——这是很多 IoT 系统的标准姿势。但真要把两个数据源塞进同一个 Spring Boot 应用配置、注入、防注入、分页、健康检查每一步都有坑。本文用实战源码讲透双数据源查询层的完整实现。数据写进去了怎么查出来前几篇我们把 MQTT 消息落进了 TDengine 超级表设备档案存进了 PostgreSQL。现在问题来了一个 Spring Boot 应用怎么同时连两个数据库你可能会想搞个分布式事务引入中间件都不需要。查询层的做法是让两个数据源各司其职服务层编排。今天这篇我们就从零拆解java-api/模块的双数据源实现。双数据源配置4 个 Bean 打天下先看核心配置类DataSourceConfiguration.java。这里定义了 4 个 Bean两个 DataSource两个 JdbcTemplate。**档案数据源PG**用 Hikari 连接池标了PrimaryBean(metadataDataSource)PrimaryConfigurationProperties(spring.datasource.hikari)publicHikariDataSourcemetadataDataSource(Qualifier(metadataDataSourceProperties)DataSourcePropertiesproperties){returnproperties.initializeDataSourceBuilder().type(HikariDataSource.class).build();}**时序数据源TDengine**用的是DriverManagerDataSource每次新建连接不给连接池Bean(tdengineDataSource)publicDataSourcetdengineDataSource(TdenginePropertiesproperties){DriverManagerDataSourcedataSourcenewDriverManagerDataSource();dataSource.setDriverClassName(com.taosdata.jdbc.ws.WebSocketDriver);dataSource.setUrl(properties.url());dataSource.setUsername(properties.username());dataSource.setPassword(properties.password());returndataSource;}注意这个驱动com.taosdata.jdbc.ws.WebSocketDriver。它走的是TAOS-WS 协议连的是 6041 端口——正是第 6 篇里那个 WebSocket 门。这意味着 Java 侧不需要装本地客户端一个 JDBC 驱动就搞定。URL 长这样jdbc:TAOS-WS://localhost:6041/iot?userrootpasswordtaosdata两个 JdbcTemplate 也对应建好metadataJdbcTemplate和tdengineJdbcTemplate。前者查档案后者查时序。注入时用Qualifier区分——比如 Repository 里是构造器注入publicTelemetryRepository(Qualifier(tdengineJdbcTemplate)JdbcTemplatejdbc){this.jdbcjdbc;}TDengine 侧没配连接池DriverManagerDataSource每次查询新建连接。查询型接口的负载可控时这样最省事连接数压上去再换 Hikari 也不迟。时序模板额外设了两件事setQueryTimeout(30)慢查询 30 秒掐断和setFetchSize(1000)流式取数防止大结果集撑爆内存。为什么档案和时序必须分开有人问都放一个库里不行吗看数据特征就明白了。PG 里的 device 表id、display_name、product_key、factory_id、workshop_id、device_type、region、enabled、metadata JSONB。这是档案数据——小、低频变更、事务性强。改名、停用、改 metadata都是典型的关系库操作。TDengine 里的 telemetry 超级表温度、湿度、电压、电流……这是时序数据——大、只追加、按时间窗口聚合。用 TDengine 的超级表 窗口函数一条 SQL 就能算小时均值。关键约束不跨库 join。档案在 PG时序在 TDengine一个查询里拿不到两张表。怎么办服务层编排。比如查询设备最新状态先从 PG 查设备档案再从 TDengine 查最新遥测点服务层把两者拼装成DeviceLatest返回查询白名单列名和 INTERVAL 怎么防注入这是本篇的一个关键点。看这段聚合 SQLStringsql SELECT _wstart AS window_start, AVG(%s) AS average_value, ... FROM iot.telemetry WHERE device_id ? AND ts ? AND ts ? INTERVAL(%s) .formatted(safeMetric,safeMetric,safeMetric,safeMetric,safeInterval);注意聚合列名和 INTERVAL 是拼进去的但设备 ID 和时间范围全用?参数绑定。为什么列名不能参数绑定因为 SQL 语法上列名和窗口函数参数不是值?占位符在这里不生效。你没法写AVG(?)让 JDBC 帮你填列名。那怎么办白名单校验。privatestaticfinalSetStringMETRICSSet.of(temperature,humidity,voltage,current_value,power,pressure,flow_rate,rotational_speed,vibration);METRICS 白名单 9 项INTERVALS 白名单 7 项10s / 30s / 1m / 5m / 15m / 1h / 1d请求里的 metric 参数先进requireMetric()校验不在白名单直接抛IllegalArgumentException。校验通过后才用.formatted()拼进 SQL。设备 ID 和时间范围呢全部?绑定。Timestamp.from(start)直接传参JDBC 驱动处理转义。这就是双保险能参数绑定的绝不拼字符串不能绑定的用白名单锁死。limit1 分页多查一条比多查一页便宜分页查询有个经典问题怎么知道还有没有下一页常规做法是再查一次 count。这里用的是limit1ListTelemetryPointrowsrepository.findTelemetry(deviceId,start,end,safeOffset,safeLimit1);returnPageResponse.of(rows,safeOffset,safeLimit);查limit 1条如果返回的行数大于请求的 limit说明还有更多数据booleanhasMorerows.size()requestedLimit;ListTitemshasMore?rows.subList(0,requestedLimit):rows;多查一条比多查一页便宜——省掉一次 count 查询代价只是一条记录的传输。PageResponse四个字段items / offset / limit / hasMore。配套的QueryRangeValidator做参数校验start/end 必填且 start end跨度 ≤ maxRangeDays默认 31 天limit 1~5000offset ≥ 0违规直接抛InvalidQueryException。在线状态5 分钟阈值 HAVING 过滤设备在线怎么判定看最新一条遥测的时间戳。privatestaticfinalDurationONLINE_THRESHOLDDuration.ofMinutes(5);booleanonlinepoint!nullpoint.timestamp().isAfter(Instant.now().minus(ONLINE_THRESHOLD));最新点时间在 now-5min 内算在线否则离线。阈值固定 5 分钟这是业务规则写死在常量里。离线设备列表呢看这条 SQLSELECTdevice_idFROMiot.telemetryGROUPBYdevice_idHAVINGLAST(ts)?LIMIT?注意是 HAVING 不是 WHERE。为什么WHERE 在分组前执行这时候 LAST(ts) 还没算出来。HAVING 在分组后过滤才能用聚合函数的计算结果。/api/operations/offline-devices?minutes30limit这个接口minutes 参数限制 1~4320030 天防止有人传个负数或超大值把 TDengine 压垮。PG 档案 CRUDJSONB 和 RETURNING 的妙用档案侧的几个细节值得说。metadata 是 JSONB 类型插入时用 ObjectMapper 序列化后 CASTINSERTINTOdevice(...)VALUES(?,...,CAST(?ASjsonb))RETURNING*RETURNING 一步拿回更新后的行UPDATEdeviceSET...WHEREid?RETURNING*如果返回 null说明设备不存在抛NotFoundException。省了一次 select。可选过滤用CAST(? AS VARCHAR) IS NULL技巧WHERE(CAST(?ASVARCHAR)ISNULLORfactory_id?)参数为 null 时条件恒真一个 SQL 搞定可选过滤不用动态拼 SQL。异常处理上DuplicateKeyException转成ConflictException返回 409 语义。健康检查与可运维性TDengine 挂了API 不能挂——这是设计原则。TdengineHealthIndicator做探针SELECTSERVER_VERSION()成功即 UP异常 DOWN。Kubernetes 的 liveness/readiness 探针可以直接用这个端点。其他配置server.port 默认 8080JAVA_API_PORT环境变量可覆盖shutdown: graceful30 秒排空compression 开启management 暴露 health/info/prometheus/metricsspringdoc 开启/swagger-ui.html 在线文档接口全景与双库协作边界最后过一遍 8 个 Controller 的接口清单方法路径说明POST /api/devices创建设备metadata 任意 JSONGET /api/devices?factoryIddeviceTypeoffsetlimit分页列表可选过滤GET /api/devices/{id}档案详情PUT /api/devices/{id}更新含 enabled 停用GET /api/devices/{deviceId}/latest最新遥测 online 布尔GET /api/devices/{deviceId}/telemetry?startendoffsetlimit分页遥测GET /api/devices/{deviceId}/aggregate?metricintervalstartend窗口聚合GET /api/factories/{factoryId}/statistics?metricstartend工厂级汇总GET /api/vehicles/{vehicleId}/track?startendoffsetlimit车辆轨迹分页GET /api/operations/offline-devices?minuteslimit离线设备列表注意 alarm-rules 相关接口只列了名那是第 9 篇的素材——告警引擎会展开讲。双库协作的边界不 join、不跨库事务、服务层编排。档案查 PG时序查 TDengine数据在 Service 层拼装。回顾一下双数据源查询层的核心就这么几件事两个 DataSource 两个 JdbcTemplatePrimary Qualifier 区分白名单防注入能参数绑定的全绑定不能绑定的锁白名单limit1 分页多查一条判断 hasMore在线判定最新点 5 分钟阈值离线用 HAVING 聚合后过滤健康检查TDengine 挂了 API 照样跑你在实际项目里遇到过双数据源的坑吗比如连接池耗尽、事务失效、或者 SQL 注入的隐患欢迎在评论区聊聊你的方案。觉得有用点个关注持续获取优质内容。
档案放 PostgreSQL时序放 TDengine——这是很多 IoT 系统的标准姿势。但真要把两个数据源塞进同一个 Spring Boot 应用配置、注入、防注入、分页、健康检查每一步都有坑。本文用实战源码讲透双数据源查询层的完整实现。数据写进去了怎么查出来前几篇我们把 MQTT 消息落进了 TDengine 超级表设备档案存进了 PostgreSQL。现在问题来了一个 Spring Boot 应用怎么同时连两个数据库你可能会想搞个分布式事务引入中间件都不需要。查询层的做法是让两个数据源各司其职服务层编排。今天这篇我们就从零拆解java-api/模块的双数据源实现。双数据源配置4 个 Bean 打天下先看核心配置类DataSourceConfiguration.java。这里定义了 4 个 Bean两个 DataSource两个 JdbcTemplate。**档案数据源PG**用 Hikari 连接池标了PrimaryBean(metadataDataSource)PrimaryConfigurationProperties(spring.datasource.hikari)publicHikariDataSourcemetadataDataSource(Qualifier(metadataDataSourceProperties)DataSourcePropertiesproperties){returnproperties.initializeDataSourceBuilder().type(HikariDataSource.class).build();}**时序数据源TDengine**用的是DriverManagerDataSource每次新建连接不给连接池Bean(tdengineDataSource)publicDataSourcetdengineDataSource(TdenginePropertiesproperties){DriverManagerDataSourcedataSourcenewDriverManagerDataSource();dataSource.setDriverClassName(com.taosdata.jdbc.ws.WebSocketDriver);dataSource.setUrl(properties.url());dataSource.setUsername(properties.username());dataSource.setPassword(properties.password());returndataSource;}注意这个驱动com.taosdata.jdbc.ws.WebSocketDriver。它走的是TAOS-WS 协议连的是 6041 端口——正是第 6 篇里那个 WebSocket 门。这意味着 Java 侧不需要装本地客户端一个 JDBC 驱动就搞定。URL 长这样jdbc:TAOS-WS://localhost:6041/iot?userrootpasswordtaosdata两个 JdbcTemplate 也对应建好metadataJdbcTemplate和tdengineJdbcTemplate。前者查档案后者查时序。注入时用Qualifier区分——比如 Repository 里是构造器注入publicTelemetryRepository(Qualifier(tdengineJdbcTemplate)JdbcTemplatejdbc){this.jdbcjdbc;}TDengine 侧没配连接池DriverManagerDataSource每次查询新建连接。查询型接口的负载可控时这样最省事连接数压上去再换 Hikari 也不迟。时序模板额外设了两件事setQueryTimeout(30)慢查询 30 秒掐断和setFetchSize(1000)流式取数防止大结果集撑爆内存。为什么档案和时序必须分开有人问都放一个库里不行吗看数据特征就明白了。PG 里的 device 表id、display_name、product_key、factory_id、workshop_id、device_type、region、enabled、metadata JSONB。这是档案数据——小、低频变更、事务性强。改名、停用、改 metadata都是典型的关系库操作。TDengine 里的 telemetry 超级表温度、湿度、电压、电流……这是时序数据——大、只追加、按时间窗口聚合。用 TDengine 的超级表 窗口函数一条 SQL 就能算小时均值。关键约束不跨库 join。档案在 PG时序在 TDengine一个查询里拿不到两张表。怎么办服务层编排。比如查询设备最新状态先从 PG 查设备档案再从 TDengine 查最新遥测点服务层把两者拼装成DeviceLatest返回查询白名单列名和 INTERVAL 怎么防注入这是本篇的一个关键点。看这段聚合 SQLStringsql SELECT _wstart AS window_start, AVG(%s) AS average_value, ... FROM iot.telemetry WHERE device_id ? AND ts ? AND ts ? INTERVAL(%s) .formatted(safeMetric,safeMetric,safeMetric,safeMetric,safeInterval);注意聚合列名和 INTERVAL 是拼进去的但设备 ID 和时间范围全用?参数绑定。为什么列名不能参数绑定因为 SQL 语法上列名和窗口函数参数不是值?占位符在这里不生效。你没法写AVG(?)让 JDBC 帮你填列名。那怎么办白名单校验。privatestaticfinalSetStringMETRICSSet.of(temperature,humidity,voltage,current_value,power,pressure,flow_rate,rotational_speed,vibration);METRICS 白名单 9 项INTERVALS 白名单 7 项10s / 30s / 1m / 5m / 15m / 1h / 1d请求里的 metric 参数先进requireMetric()校验不在白名单直接抛IllegalArgumentException。校验通过后才用.formatted()拼进 SQL。设备 ID 和时间范围呢全部?绑定。Timestamp.from(start)直接传参JDBC 驱动处理转义。这就是双保险能参数绑定的绝不拼字符串不能绑定的用白名单锁死。limit1 分页多查一条比多查一页便宜分页查询有个经典问题怎么知道还有没有下一页常规做法是再查一次 count。这里用的是limit1ListTelemetryPointrowsrepository.findTelemetry(deviceId,start,end,safeOffset,safeLimit1);returnPageResponse.of(rows,safeOffset,safeLimit);查limit 1条如果返回的行数大于请求的 limit说明还有更多数据booleanhasMorerows.size()requestedLimit;ListTitemshasMore?rows.subList(0,requestedLimit):rows;多查一条比多查一页便宜——省掉一次 count 查询代价只是一条记录的传输。PageResponse四个字段items / offset / limit / hasMore。配套的QueryRangeValidator做参数校验start/end 必填且 start end跨度 ≤ maxRangeDays默认 31 天limit 1~5000offset ≥ 0违规直接抛InvalidQueryException。在线状态5 分钟阈值 HAVING 过滤设备在线怎么判定看最新一条遥测的时间戳。privatestaticfinalDurationONLINE_THRESHOLDDuration.ofMinutes(5);booleanonlinepoint!nullpoint.timestamp().isAfter(Instant.now().minus(ONLINE_THRESHOLD));最新点时间在 now-5min 内算在线否则离线。阈值固定 5 分钟这是业务规则写死在常量里。离线设备列表呢看这条 SQLSELECTdevice_idFROMiot.telemetryGROUPBYdevice_idHAVINGLAST(ts)?LIMIT?注意是 HAVING 不是 WHERE。为什么WHERE 在分组前执行这时候 LAST(ts) 还没算出来。HAVING 在分组后过滤才能用聚合函数的计算结果。/api/operations/offline-devices?minutes30limit这个接口minutes 参数限制 1~4320030 天防止有人传个负数或超大值把 TDengine 压垮。PG 档案 CRUDJSONB 和 RETURNING 的妙用档案侧的几个细节值得说。metadata 是 JSONB 类型插入时用 ObjectMapper 序列化后 CASTINSERTINTOdevice(...)VALUES(?,...,CAST(?ASjsonb))RETURNING*RETURNING 一步拿回更新后的行UPDATEdeviceSET...WHEREid?RETURNING*如果返回 null说明设备不存在抛NotFoundException。省了一次 select。可选过滤用CAST(? AS VARCHAR) IS NULL技巧WHERE(CAST(?ASVARCHAR)ISNULLORfactory_id?)参数为 null 时条件恒真一个 SQL 搞定可选过滤不用动态拼 SQL。异常处理上DuplicateKeyException转成ConflictException返回 409 语义。健康检查与可运维性TDengine 挂了API 不能挂——这是设计原则。TdengineHealthIndicator做探针SELECTSERVER_VERSION()成功即 UP异常 DOWN。Kubernetes 的 liveness/readiness 探针可以直接用这个端点。其他配置server.port 默认 8080JAVA_API_PORT环境变量可覆盖shutdown: graceful30 秒排空compression 开启management 暴露 health/info/prometheus/metricsspringdoc 开启/swagger-ui.html 在线文档接口全景与双库协作边界最后过一遍 8 个 Controller 的接口清单方法路径说明POST /api/devices创建设备metadata 任意 JSONGET /api/devices?factoryIddeviceTypeoffsetlimit分页列表可选过滤GET /api/devices/{id}档案详情PUT /api/devices/{id}更新含 enabled 停用GET /api/devices/{deviceId}/latest最新遥测 online 布尔GET /api/devices/{deviceId}/telemetry?startendoffsetlimit分页遥测GET /api/devices/{deviceId}/aggregate?metricintervalstartend窗口聚合GET /api/factories/{factoryId}/statistics?metricstartend工厂级汇总GET /api/vehicles/{vehicleId}/track?startendoffsetlimit车辆轨迹分页GET /api/operations/offline-devices?minuteslimit离线设备列表注意 alarm-rules 相关接口只列了名那是第 9 篇的素材——告警引擎会展开讲。双库协作的边界不 join、不跨库事务、服务层编排。档案查 PG时序查 TDengine数据在 Service 层拼装。回顾一下双数据源查询层的核心就这么几件事两个 DataSource 两个 JdbcTemplatePrimary Qualifier 区分白名单防注入能参数绑定的全绑定不能绑定的锁白名单limit1 分页多查一条判断 hasMore在线判定最新点 5 分钟阈值离线用 HAVING 聚合后过滤健康检查TDengine 挂了 API 照样跑你在实际项目里遇到过双数据源的坑吗比如连接池耗尽、事务失效、或者 SQL 注入的隐患欢迎在评论区聊聊你的方案。觉得有用点个关注持续获取优质内容。
