四合一加湿除湿消毒净化一体机接入腾讯云 IoT Explorer：MQTT 上报 + 规则引擎告警实操

原创

盛世宏博科技

发布于 2026-09-08 16:34:34

1870

四合一环境一体机接入腾讯云 IoT Explorer：MQTT 上报 + 规则引擎告警实操

设备侧：四合一一体机（UVC 消毒日志 + 温湿度 + PM2.5 + 湿度控制状态），固件已开放 MQTT 客户端配置。本文记录我们把样机从零挂到 IoT Explorer、用规则引擎做越限推送、再挂一条 SCF 云函数推企业微信的完整链路。开发环境 Ubuntu 22.04，工具 mosquitto_pub + 腾讯云 CLI。

一、设备侧数据模型先定清楚

一体机每 5s 往主题推一帧 JSON，避免把四个子系统拆四个 Topic 造成影子碎片：

{
  "dev_id": "AQ-04IN1-007",
  "ts": 1788850000,
  "temp": 22.4,
  "rh": 53.8,
  "pm25": 8,
  "voc_idx": 12,
  "uvc_dose_mj": 142.6,
  "uvc_last_sec": 180,
  "mode": "auto",
  "fan_rpm": 1180,
  "water_tray_ok": true
}

上报主题约定：

$thing/up/property/{ProductID}/{DeviceName}

下发控制（切消毒模式 / 改湿度设定点）走：

$thing/down/control/{ProductID}/{DeviceName}

二、IoT Explorer 侧建模（3 步）

1. 建产品 → 选择"环境监测设备"品类，认证方式 PSK-Cert 二选一，我们选 X.509 证书（合规项目别用 PSK）

数据模板里挂以下属性：

标识符	类型	读写	备注
temp	float	只读	±0.3℃ 校准值
rh	float	只读	恒湿闭环反馈值
pm25	int	只读	激光粒子计
uvc_dose_mj	float	只读	累计消毒剂量，审计用
uvc_last_sec	int	只读	上次消毒时长
mode	enum	读写	auto/manual/sterilize
rh_setpoint	float	读写	下发湿度设定点
fan_rpm	int	只读	风档反馈

2. 启设备影子（Device Shadow）——断网期间一体机本地缓存的 30 天数据补传时，影子做最后已知值兜底，避免平台侧出现空窗。

3. 拿到三元组：ProductID / DeviceName / 设备证书 PEM + 私钥 PEM，塞进一体机 Web 配置页的 MQTT 客户端栏，Broker 填：

mqtts://{ProductID}.iotcloud.tencent.com:8883

三、规则引擎：越限不写一行后端

不用自己搭消费者服务，直接在 Explorer → 规则引擎 → 新建 SQL：

SELECT
  dev_id, ts, temp, rh, pm25, uvc_last_sec
FROM
  "$thing/up/property/{{ProductID}}/+"
WHERE
  rh > 60
  OR rh < 40
  OR pm25 > 35
  OR temp NOT BETWEEN 18 AND 26

动作挂两个：

动作 A：转发到 CMQ Topic → 触发 SCF 云函数（下节）
动作 B：直接推 CKafka，留一份原始审计流给合规库

💡 经验值：湿度带别写 rh <> [40,60] 这种，现场偶尔 1 帧抖动会刷爆告警。我们在 WHERE 外再叠一层平台侧"连续 3 次越限才触发"的防抖窗口（Explorer 规则引擎支持 LAG() 风格窗口函数，没有的话在 SCF 里做滑动计数也行）。

四、SCF 云函数：推企业微信 + 联动消毒强启

规则命中后 CMQ 触发 SCF（Python 3.10 运行时），逻辑很薄：

# index.py
import json
import requests

WX_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx"
SHADOW_API = "https://iotexplorer.tencentcloudapi.com/"

def main_handler(event, context):
    for rec in event.get("Records", []):
        payload = json.loads(rec["body"])
        d = payload.get("d", {})
        dev = d.get("dev_id", "unknown")

        # 组装企微卡片
        md = (
            f"### ⚠️ 环境越限告警\n"
            f"> 设备：**{dev}**\n"
            f"> 温度：{d.get('temp')}℃ / 湿度：{d.get('rh')}%\n"
            f"> PM2.5：{d.get('pm25')} μg/m³\n"
        )
        # 湿度越上限 → 强制切消毒 + 降湿设定点
        if d.get("rh", 0) > 60:
            _shadow_control(dev, {"mode": "sterilize", "rh_setpoint": 45.0})
            md += "> 处置：**已下发 UVC 强启 + 湿度回设 45%RH**\n"

        requests.post(WX_WEBHOOK, json={
            "msgtype": "markdown", "markdown": {"content": md}
        }, timeout=3)
    return {"ok": True}

def _shadow_control(dev, patch):
    # 走 Explorer 设备影子下发，断网设备恢复后自动执行
    # 实际调用 TencentCloud SDK ModifyDeviceShadow
    pass

部署：

tccli scf CreateFunction \
  --FunctionName env-alert-hook \
  --Runtime Python3.10 \
  --Handler index.main_handler \
  --Code ZipFile=$(base64 -w0 dist.zip)

五、断网补传的影子行为（踩坑记录）

一体机在弱电间偶尔断网 10–40 分钟，恢复后按本地 Seq 补推 5s 粒度帧。我们验证过三种行为：

场景	裸 MQTT 无影子	开设备影子
断网期间平台侧曲线	出现空窗，合规审计标记缺口	影子保留最后已知值，补传帧到达后回填
补传时间戳	需自己写服务端去重	Explorer 按 ts 字段自动归位，无重复点
下发指令在断网期发出	丢失	影子缓存，设备上线即执行

结论：合规类项目（GSP/SEMI/等保）影子必须开，不要图省事裸 Pub/Sub。

六、压测与验证数据（我们实测的，不是参数页话术）

单产品下并发 200 台一体机，5s 上报周期，规则引擎 P99 延迟 87ms
断网 30 min 模拟补传，单设备回填 360 帧，影子无丢点
企微告警在越限第 3 帧触发（防抖窗口生效），误报率为 0 / 72h 试运行

七、收尾：给后来人的三条

证书别硬编码在固件镜像里，让现场用 Web 页导入 PEM，过等保时少一道整改单
规则 SQL 的 WHERE 一定加防抖，否则 PM 传感器偶发毛刺会刷爆值班手机
消毒剂量 uvc_dose_mj 走独立审计 Topic 单独留 Kafka，别和实时控制流混一个消费组，飞检时调日志能秒出

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

开发经验

教程

腾讯云测试服务