摘要

高速公路隧道是交通安全的高风险区域，对 CO 浓度、能见度、风速等环境指标的实时监控直接关系到通行安全。传统方案依赖外部流处理框架（如 Flink、Spark Streaming），架构复杂且延迟较高。本文深入探讨如何利用时序数据库 TDengine 内置的流计算引擎，在高速公路隧道监控场景中实现秒级环境异常检测与自动告警，涵盖流式 SQL 语法、窗口策略选型与性能调优的完整实践。

---

一、隧道监控的数据挑战

1.1 监控指标体系

一座标准的高速公路隧道（长度 2km 以上）通常部署以下监测设备：

• CO/VI 检测器：每 30 秒上报一氧化碳浓度（ppm）和能见度（m）
• 风速风向仪：每 10 秒上报纵向风速和横向风速
• 亮度检测器：每 60 秒上报洞内外亮度差
• 温湿度传感器：每 60 秒上报温度和湿度
• 交通检测器：每 5 秒上报车流量、车速、占有率

一座 3 公里长的隧道，传感器数量在 200-400 个之间，每秒产生 2000-5000 条数据。一个省级隧道管理中心管辖 50-100 座隧道，合计每秒写入量达 10-50 万条。

1.2 实时性要求

隧道环境监控对告警延迟的要求极其严格：

告警级别	触发条件	最大延迟
一级（紧急）	CO > 250ppm 或能见度 < 5m	≤ 3 秒
二级（严重）	CO > 150ppm 或能见度 < 10m	≤ 10 秒
三级（警告）	CO > 100ppm 或能见度 < 15m	≤ 30 秒

传统架构需要 TDengine + Kafka + Flink + 告警服务，链路长、延迟高、运维复杂。

---

二、TDengine 流计算架构

2.1 为什么选择内置流计算？

TDengine 从 3.0 版本开始内置流计算引擎，相比外部流处理框架，具有三大优势：

1. 零额外组件：无需部署 Kafka、Flink 等中间件，数据写入即触发计算
2. SQL 原生语法：使用标准 SQL 定义流计算逻辑，学习成本极低
3. 存算一体：计算结果直接写入 TDengine 表，避免跨系统数据搬运

2.2 隧道监控数据模型

-- 创建隧道环境监测超级表
CREATE STABLE IF NOT EXISTS tunnel.env_sensor (
    ts           TIMESTAMP,
    co_ppm       FLOAT,          -- CO 浓度（ppm）
    visibility   FLOAT,          -- 能见度（m）
    wind_speed   FLOAT,          -- 纵向风速（m/s）
    temperature  FLOAT,          -- 温度（℃）
    humidity     FLOAT           -- 湿度（%）
) TAGS (
    tunnel_id    BINARY(30),     -- 隧道编号
    sensor_id    BINARY(30),     -- 传感器编号
    zone         BINARY(10),     -- 区段（入口/中段/出口）
    stake_mark   BINARY(20)      -- 桩号
);

-- 创建交通检测超级表
CREATE STABLE IF NOT EXISTS tunnel.traffic_detector (
    ts           TIMESTAMP,
    flow         INT,            -- 车流量（辆/周期）
    speed        FLOAT,          -- 平均车速（km/h）
    occupancy    FLOAT           -- 占有率（%）
) TAGS (
    tunnel_id    BINARY(30),
    detector_id  BINARY(30),
    lane_no      TINYINT,
    direction    TINYINT
);

---

三、流计算实战

3.1 环境异常实时检测流

-- 创建 CO/能见度异常检测流
-- 每 10 秒滑动窗口，检测超标情况
CREATE STREAM IF NOT EXISTS tunnel.stream_env_alert 
    INTO tunnel.env_alert_result
    TAGS (tunnel_id, zone)
    AS
SELECT 
    _wstart as window_start,
    tunnel_id,
    zone,
    AVG(co_ppm) as avg_co,
    MAX(co_ppm) as max_co,
    AVG(visibility) as avg_visibility,
    MIN(visibility) as min_visibility,
    COUNT(*) as sample_count,
    CASE 
        WHEN MAX(co_ppm) > 250 OR MIN(visibility) < 5 THEN 1
        WHEN MAX(co_ppm) > 150 OR MIN(visibility) < 10 THEN 2
        WHEN MAX(co_ppm) > 100 OR MIN(visibility) < 15 THEN 3
        ELSE 0
    END as alert_level
FROM tunnel.env_sensor
INTERVAL(10s) SLIDING(5s)
HAVING alert_level > 0;

3.2 交通拥堵检测流

-- 创建隧道交通拥堵检测流
-- 当车速低于 20km/h 且占有率超过 80%，判定为拥堵
CREATE STREAM IF NOT EXISTS tunnel.stream_traffic_jam
    INTO tunnel.traffic_jam_result
    TAGS (tunnel_id, direction)
    AS
SELECT
    _wstart as window_start,
    tunnel_id,
    direction,
    AVG(speed) as avg_speed,
    AVG(occupancy) as avg_occupancy,
    SUM(flow) as total_flow
FROM tunnel.traffic_detector
INTERVAL(30s)
HAVING avg_speed < 20 AND avg_occupancy > 80;

3.3 联动风机控制策略

-- 基于 CO 浓度和风速的风机联动策略流
CREATE STREAM IF NOT EXISTS tunnel.stream_fan_control
    INTO tunnel.fan_control_cmd
    TAGS (tunnel_id, zone)
    AS
SELECT
    _wstart as window_start,
    tunnel_id,
    zone,
    AVG(co_ppm) as avg_co,
    AVG(wind_speed) as avg_wind,
    CASE
        WHEN AVG(co_ppm) > 200 AND AVG(wind_speed) < 2 THEN 'FULL_SPEED'
        WHEN AVG(co_ppm) > 100 AND AVG(wind_speed) < 3 THEN 'HALF_SPEED'
        WHEN AVG(co_ppm) < 50 AND AVG(wind_speed) > 3 THEN 'STOP'
        ELSE 'MAINTAIN'
    END as fan_command
FROM tunnel.env_sensor
INTERVAL(30s);

---

四、告警消费与推送

流计算的结果写入 TDengine 表后，通过数据订阅（Topic）机制推送到告警服务：

import taos
from taos.tmq import Consumer

# 创建告警主题
conn = taos.connect(host="localhost", database="tunnel")
conn.execute("""
    CREATE TOPIC IF NOT EXISTS topic_env_alert 
    AS SELECT * FROM tunnel.env_alert_result 
    WHERE alert_level <= 2
""")

# 消费告警数据
consumer = Consumer({
    "group.id": "alert_service_01",
    "td.connect.ip": "localhost",
    "auto.offset.reset": "latest"
})
consumer.subscribe(["topic_env_alert"])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg:
        for row in msg:
            alert_level = row["alert_level"]
            tunnel_id = row["tunnel_id"]
            max_co = row["max_co"]
            min_vis = row["min_visibility"]
            
            if alert_level == 1:
                # 一级告警：短信 + 电话 + 大屏弹窗
                send_emergency_alert(tunnel_id, max_co, min_vis)
            elif alert_level == 2:
                # 二级告警：短信 + 大屏提示
                send_warning_alert(tunnel_id, max_co, min_vis)

consumer.close()

---

五、性能调优实践

5.1 窗口策略选型

窗口类型	适用场景	计算开销
滚动窗口 INTERVAL(10s)	定期统计、报表	低
滑动窗口 INTERVAL(10s) SLIDING(5s)	实时异常检测	中
会话窗口 SESSION(30s)	事件聚类分析	高

建议：一级告警使用滑动窗口（低延迟），统计报表使用滚动窗口（低开销）。

5.2 流计算资源配置

-- 调整流计算线程数（默认 4，高负载可增加）
ALTER DNODE 1 'streamThreads' '8';

-- 查看流计算运行状态
SELECT * FROM information_schema.ins_stream_tasks;

-- 监控流计算延迟
SELECT stream_name, sink_delay, source_delay 
FROM information_schema.ins_stream_tasks
WHERE sink_delay > 1000;  -- 延迟超过 1 秒的流

5.3 写入与流计算并行调优

在隧道监控场景中，写入和流计算共享集群资源。推荐配置：

# taos.cfg 关键参数
numOfVnodes          16          # VNode 数量
streamThreads        8           # 流计算线程
ratioOfQueryCores    0.5         # 查询核心占比
maxStreamCompDelay   5000        # 最大流计算延迟（ms）

实测在 3 节点集群下，支撑 80 座隧道（3 万传感器）的写入 + 流计算，端到端告警延迟控制在 2 秒以内。

---

六、方案对比

维度	传统方案（TDengine+Flink）	TDengine 内置流计算
组件数量	5+（DB+Kafka+Flink+Redis+告警）	2（TDengine+告警服务）
部署复杂度	高（多集群运维）	低（单一集群）
端到端延迟	5-15 秒	1-3 秒
学习成本	高（Java/Scala + Flink API）	低（标准 SQL）
资源消耗	高（多组件内存开销）	低（存算一体）

---

七、总结

TDengine 内置流计算引擎将时序数据的存储与实时计算融为一体，从根本上简化了隧道监控系统的技术架构。通过标准 SQL 定义流计算逻辑，结合数据订阅机制实现告警推送，将端到端延迟从传统方案的 10 秒级压缩到 2 秒以内，同时大幅降低了系统的部署和运维成本。

对于高速公路隧道监控场景，TDengine 的流计算方案提供了一条"少即是多"的技术路径——更少的组件，更低的延迟，更简的运维。