一条温度读数从设备上报到空调切换制冷模式，端到端只要 3 秒。这不是压测环境下的理想值，是江苏正元数币「结合智联」平台在真实校园里的运行结果。背后撑住这条链路的，是 TDengine 的流计算引擎。
先看一条数据怎么跑完全程
14:32:05，某教室温控器上报：室温 36.2℃，制冷模式，设定温度 26℃。
同一条写入链路里，TDengine 流计算引擎立刻判断 room_temperature > 35 成立，触发告警规则 std 000001。
14:32:06，告警记录落入 device_alarm_record 超级表。Scene Model Trigger Processor 从 Redis 拉取该房间的场景策略缓存，命中「连续超温 → 强制制冷 + 通知管理员」。
14:32:07，通过 Redis Pipeline 批量取回关联设备信息，组装设备命令，经 Kafka 下发到设备命令执行微服务。
14:32:08，空调收到指令，模式切为强制制冷，风速拉满。
整条链路没有定时任务在中间轮询，没有应用层反复扫表。数据落盘的那一刻，判定就已经做完了。
流计算改掉的不只是延迟
把预警判定从应用层搬进写入链路，表面看是快了几秒，实际改的是整套架构的分工方式。
原来的做法是「先存后算」：数据全部入库，Java 定时任务每隔一段时间扫一遍，发现越限再写告警。数据被搬来搬去，延迟完全取决于轮询间隔。
CREATE STREAM 把判定逻辑直接注入写入链路，数据落盘前就过一遍流计算。命中规则的数据写入告警表，正常数据照常入库。存储和判定在同一条链路里完成，没有搬运，没有等待。
窗口类型的选择决定了误报率。COUNT_WINDOW 处理「来一条判一条」的瞬时预警，EVENT_WINDOW 处理「持续超限才告警」的持续性判定。两者配合 TRUE_FOR 抑制抖动误报，再用 HAVING COUNT(*) >= 10 确保联动触发依赖的是持续趋势，而不是某一次孤立读数。
性能层面靠三个设计兜底：PRE_FILTER 在流计算入口就把正常数据滤掉，不浪费算力；PARTITION BY tbname 让每台设备独立分区并行计算，告警延迟不随设备数量增长而恶化；告警结果写入独立超级表，跟业务数据的查询写入互不干扰。
对开发来说，告警规则就是 SQL。新增规则建一条 STREAM，调整阈值改 PRE_FILTER 条件，不用改 Java 代码，不用重新发版。应用层从异常判定里彻底解放出来，只负责消费告警事件、匹配策略、下发指令。
为什么关联库加定时任务撑不住
「结合智联」早期版本用的是标准打法：设备数据写关系型数据库，每台设备一张表；预警靠 Java 定时任务每 30 秒扫一次近 1 分钟的数据；联动控制由独立的规则引擎轮询设备状态表来判断触发条件。
设备量从几百涨到几万之后，问题集中爆发。
写入侧，万台设备每秒各 1 条就是 10000 QPS 的持续写入，关系库的写入队列一直堆积。告警侧，30 秒的定时间隔加上 SQL 扫描本身的耗时，实际延迟经常超过 2 分钟。联动侧，规则引擎轮询设备状态表，高并发下锁竞争严重。存储侧，3 个月历史数据超过 2TB，按月聚合查询要跑好几分钟。
换成 TDengine 之后，三个层面的变化最关键。
数据模型上，一设备一子表，超级表统一管理，TAG 标记组织、空间、设备 ID。不用再为每台设备手动建表，查询也不用 UNION 几百张表。
计算位置上，CREATE STREAM 把预警判定注入写入链路，数据落盘前先过流计算，命中规则直接写告警表。应用层不再轮询，只需订阅告警表的变化。
联动架构上，从定时轮询变成事件驱动：告警产生 → 应用层消费告警事件 → 匹配场景策略 → 下发控制指令。
这里有个坑值得提醒：流计算不是银弹。如果你的场景是低频写入、规则经常全量变更、或者需要跨大量历史数据做复杂关联分析，传统批处理反而更合适。流计算的优势区间是高频写入 + 规则相对稳定的实时判定场景。
瞬时预警：数据入库即判定
室内温度超过 35℃ 立即产生告警记录，用 COUNT_WINDOW 实现：
几个关键设计点值得展开。
count_window(1, room_temperature) 表示每来 1 条数据（只要 room_temperature 非 NULL）就触发一次判定，不攒批。PRE_FILTER(room_temperature > 35) 在流计算入口就过滤掉正常数据，只有超限数据才进入后续逻辑，避免无效计算。MAX_DELAY(10s) 容忍最多 10 秒的数据乱序，在实时性和容错之间取平衡。
MAX_DELAY 这个参数容易设错。设太小，乱序数据会被丢弃；设太大，告警延迟增加。10 秒对于校园物联网场景是合理的经验值，但具体数值要根据实际网络抖动情况调整。
持续预警：用事件窗口压住告警风暴
瞬时预警有个现实问题：温度在 35℃ 附近抖动时，会触发大量告警。对于需要持续关注的场景，得用事件窗口——只有温度持续超限 30 分钟才触发，并输出窗口内的均值温度：
CREATE STREAM IF NOT EXISTS stream_alarm_dynamic_std 000002 EVENT_WINDOW(START WITH room_temperature > 35 END WITH room_temperature <= 35) TRUE_FOR (30m) FROM device_report_info_record PARTITION BY tbname, organization_id, room_id, client_id STREAM_OPTIONS(IGNORE_DISORDER | MAX_DELAY(30m)) INTO device_alarm_record(ts, generate_time, alarm_rule_id, alarm_value) TAGS (organization_id BIGINT AS organization_id, room_id BIGINT AS room_id, client_id VARCHAR(255) AS client_id) AS SELECT CAST(_tlocaltime/1000000 AS TIMESTAMP) as ts, _twend as generate_time, CAST(“std 000002” AS VARCHAR(10)) AS alarm_rule_id, CAST(AVG(room_temperature) AS VARCHAR(255)) AS alarm_value FROM %%tbname WHERE ts >= _twstart AND ts <= _twend;
EVENT_WINDOW 的设计思路不是简单的「连续 N 条超限就告警」，而是追踪一个事件的完整生命周期：从温度突破阈值（START WITH）到回落（END WITH）。如果事件持续超过 TRUE_FOR(30m)，才触发告警并输出窗口内的均值。这样既避免了抖动误报，又给出了告警期间的统计摘要。
实测下来，事件窗口对持续性异常场景的误报抑制效果明显优于纯计数窗口。代价是告警延迟至少等于 TRUE_FOR 设定的时长，所以不适合对响应速度要求极高的场景。
用电异常：每小时用电量超标自动告警
这个场景面向受控设备（如智慧断路器）的每小时用电量做实时监测，超出规则阈值就自动产出告警记录：
CREATE STREAM IF NOT EXISTS stream_electricity_alarm_dynamic_stde 000001 count_window(1, electricity_consumption_hour) FROM controlled_device_electricity_consumption_hour PARTITION BY tbname, organization_id, room_id, controlled_device_id STREAM_OPTIONS(PRE_FILTER(electricity_consumption_hour > 5.0) | MAX_DELAY(10s)) INTO controlled_device_electricity_consumption_hour_alarm_record(ts, generate_time, electricity_alarm_rule_id, electricity_alarm_rule_value) TAGS (organization_id BIGINT AS organization_id, room_id BIGINT AS room_id, controlled_device_id VARCHAR(16) AS controlled_device_id) AS SELECT CAST(_tlocaltime/1000000 AS TIMESTAMP) as ts, _rowts as generate_time, CAST(“stde 000001” AS VARCHAR(10)) AS electricity_alarm_rule_id, CAST(electricity_consumption_hour AS VARCHAR(255)) AS electricity_alarm_rule_value FROM %%tbname;
用电异常监测对数据时效性要求很高。传统做法依赖定时任务批量扫描历史数据，告警经常滞后十几分钟甚至小时级。TDengine 流计算通过 PARTITION BY tbname 按设备子表分区并行计算，每台受控设备的用电数据写入后立即触发判定，异常告警在 10 秒内产出。
告警结果写入独立的告警超级表，与业务数据物理隔离。告警记录通过定时查询同步到 MySQL 场景库：
SELECT ts, generate_time, electricity_alarm_rule_id, electricity_alarm_rule_value, organization_id, room_id, controlled_device_id FROM controlled_device_electricity_consumption_hour_alarm_record WHERE ts >= ‘2025-07-25 00:00:00’ AND ts < ‘2025-07-25 00:01:00’
注意这里的时间范围查询。TDengine 按时间分区存储，查询时务必带上 ts 范围条件，否则会全表扫描。这是从关系库迁移过来时最容易忽略的一点。
从告警到控制：事件驱动的联动回路
告警只是手段，控制才是目的。正元数币的差异化能力在于：告警不只是推到大屏上，而是自动触发设备控制。
联动架构的链路是这样的：设备上报 → TDengine（写入 + 流计算判定）→ 告警事件 → Scene Model Trigger Processor（Java 微服务）→ Redis 缓存取场景策略 → Redis Pipeline 批量取关联设备状态 → Kafka 异步下发设备命令 → 设备命令执行微服务 → 设备执行 → 结果 Kafka 回传。
核心处理逻辑：
public void trigger(String clientId, Long roomId) {
// 1. 获取该房间下有效的场景模型策略缓存
List validStrategies = sceneModelCacheHelper.getValidSceneModelStrategyCache(roomId);
// 2. 使用 Redis 管道批量获取设备产品ID、设备ID、状态
List devicesCommandInfoList = redisTemplate.executePipelined(…);
// 3. 按产品ID映射设备，组装设备命令
validStrategies.forEach(strategy -> {
List commandList = new ArrayList<>();
strategy.getBelongSceneModelCache()
.getSceneModelProductCommands()
.forEach(productCommand -> {
List devices = productDeviceCommandInfoMap.get(productCommand.getProductId());
devices.forEach(device -> {
commandList.add(DeviceCommand.create()
.clientId(device.getClientId())
.serviceId(productCommand.getServiceId())
.command(productCommand.getCommand()));
});
});
// 4. 通过 Kafka 异步下发设备命令任务
DeviceCommandTask task = DeviceCommandTask.create()
.deviceCommandList(commandList);
kafkaTemplate.send(KafkaTopic.SEND_DEVICE_COMMAND_TASK, task);
});
}
这个架构有三个设计要点。
Redis 双层缓存。场景模型策略（哪个房间、什么条件、触发什么动作）是静态配置，缓存在 Redis 中，触发时直接读取。设备信息同样通过 Redis Pipeline 在一次网络往返中批量获取，避免逐条查询。
Kafka 削峰填谷。大面积温控异常时（比如夏季制冷系统故障），可能瞬间产生数千条告警，对应数千条控制命令。Kafka 作为异步消息中间件，确保命令不丢失、不阻塞主链路，设备执行微服务按自身消费能力拉取。
失败精准追溯。每条设备命令的执行结果通过 Kafka 回传，系统记录每台设备的执行状态（成功/失败/超时），失败设备可精确重试。
触发条件不能只看单次读数
不是所有告警都需要联动。有些场景需要依赖历史数据判断趋势，比如「房间持续低温 → 自动制热」：
SELECT DISTINCT room_id FROM device_report_info_record WHERE client_id IN (‘device_001’, ‘device_002’) AND ts >= NOW – 30m AND room_temperature < 16 GROUP BY room_id HAVING COUNT(*) >= 10;
条件是「30 分钟内同一房间至少有 10 条温度低于 16℃ 的记录」。联动触发依赖的是持续趋势，不是某一次读数。这个查询本身也可以改写成流计算任务，进一步降低延迟。
流计算之外的基础能力
除了实时预警和联动，「结合智联」平台还覆盖了绿色校园的三大核心路径。
教室智慧温控与场景联动。根据课表、作息、节假日等时间维度自动调控教室空调和照明，避免长明灯和空调空转。功能包括定时策略、场景模型建设（人体感应调控设备）、课表联动。
宿舍安全用能管控。整合大功率电器智能识别机制，检测到违规用电时及时启动电路保护。功能包括用电拆解、电控策略（最大功率、分时限流、过温跳电、欠费断电、恶性负载）、电费缴纳（电费定价、充值缴费、免费额度设置、余额提醒）。
能耗监测。实时掌握能耗构成与变化趋势，及时发现异常点。功能包括用电拆解、精确计量、用电趋势对比（同比/环比/同区域对标）、能耗报告、电控策略、异常告警。
验证流计算是否真的生效
部署完流计算任务后，别只看任务状态是 running 就放心。几个验证动作建议做一遍。
确认流计算任务已创建：SHOW STREAMS; 检查输出表是否有数据写入：SELECT COUNT(*) FROM device_alarm_record WHERE ts >= NOW – 1h; 用一条模拟数据验证端到端延迟：手动插入一条 room_temperature > 35 的记录，观察告警表多久出现对应记录。
判定标准很直接：从插入到告警表可查，延迟应该在秒级。如果超过 10 秒，检查 MAX_DELAY 设置和分区键是否合理。如果告警表一直没有数据，先确认 PRE_FILTER 条件是否写反了——这是最常见的低级错误。
