影刀RPA MQTT物联网自动化传感器数据采集与设备监控作者林焱 | 适用人群影刀RPA新手 | 难度★★★☆☆一、什么情况用这个做物联网运维的兄弟每天要盯着几十个传感器数据——温度、湿度、PM2.5、设备状态。靠人工一个个看仪表盘不光累而且容易漏掉异常。MQTT是物联网领域的标准协议大部分IoT设备都支持。但你不可能天天开着MQTT客户端盯着。这时候就需要影刀RPA出场了定时订阅MQTT主题 → 解析JSON数据 → 写Excel记录 → 超标自动告警。一套流程下来你只需要在收到告警时看一眼其余时间机器人替你盯着。实际场景工厂车间环境监控、冷链仓储温度记录、机房设备状态巡检。这篇文章能解决的问题定时从MQTT Broker拉取传感器数据自动解析JSON格式的IoT消息数据超标自动推送告警企微/钉钉生成每日设备运行报表二、怎么做2.1 准备工作影刀中创建新流程命名为「MQTT数据采集监控」。首先要安装Python依赖。影刀的Python节点需要用pip安装paho-mqtt库# 在影刀的【执行Python代码】节点中先确保安装了paho-mqtt# 如果环境受限可以用pip在影刀内置Python中安装# C:\Users\你的用户名\AppData\Local\yingdao\python\python.exe -m pip install paho-mqtt[video(video-v6OFYhUz-1783577039493)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限)]2.2 主流程设计在影刀中整个流程分5步步骤影刀操作说明①【执行Python代码】连接MQTT订阅主题接收消息②【执行Python代码】解析JSON数据提取关键字段③【写入Excel】将数据追加到当天的Excel文件④【条件判断】判断数据是否超过阈值⑤【发送企业微信消息】超标时推送告警2.3 核心Python代码步骤①连接MQTT并接收消息在影刀中拖入【执行Python代码】节点写入以下代码importpaho.mqtt.clientasmqttimportjsonimporttime# 配置区域根据实际情况修改MQTT_BROKER192.168.1.100# MQTT服务器地址MQTT_PORT1883# 端口默认1883MQTT_TOPICS[(factory/workshop1/temperature,0),# 车间1温度(factory/workshop1/humidity,0),# 车间1湿度(factory/workshop2/temperature,0),# 车间2温度]TIMEOUT30# 等待消息的超时秒数# 接收消息 received_messages[]defon_connect(client,userdata,flags,rc):连接成功的回调ifrc0:print(MQTT连接成功)fortopic,qosinMQTT_TOPICS:client.subscribe(topic,qos)else:print(f连接失败返回码{rc})defon_message(client,userdata,msg):收到消息的回调try:payloadjson.loads(msg.payload.decode(utf-8))received_messages.append({topic:msg.topic,value:payload.get(value),unit:payload.get(unit,),timestamp:payload.get(timestamp,time.time()),device_id:payload.get(device_id,),})exceptExceptionase:print(f解析消息失败{e}原始数据{msg.payload})# 创建客户端并连接clientmqtt.Client()client.on_connecton_connect client.on_messageon_message# 如果MQTT需要用户名密码取消下面注释# client.username_pw_set(admin, password)client.connect(MQTT_BROKER,MQTT_PORT,60)client.loop_start()# 等待一段时间收集消息time.sleep(TIMEOUT)client.loop_stop()client.disconnect()# 将结果存为JSON字符串供影刀后续节点使用resultjson.dumps(received_messages,ensure_asciiFalse)print(f共收到{len(received_messages)}条消息)踩坑提醒time.sleep(TIMEOUT)的时间要合理设置。太短收不到足够的数据太长流程会卡住。建议先在影刀外部用MQTT客户端测试消息频率再确定超时时间。代码执行后result变量会传给影刀你需要在影刀中用一个变量比如mqtt_data来接收。步骤②解析JSON提取数据再拖入一个【执行Python代码】节点importjson# 从上一个节点传入的JSON字符串raw_datamqtt_data# 影刀变量会自动注入messagesjson.loads(raw_data)# 整理成结构化数据records[]formsginmessages:# 从topic中提取位置信息topic_partsmsg[topic].split(/)locationtopic_parts[1]iflen(topic_parts)1elseunknownmetrictopic_parts[2]iflen(topic_parts)2elseunknownrecords.append({location:location,metric:metric,value:msg[value],unit:msg[unit],device_id:msg[device_id],timestamp:msg[timestamp],})# 转为影刀能处理的格式——返回列表的字符串表示# 影刀可以直接用【遍历列表】指令处理outputjson.dumps(records,ensure_asciiFalse)步骤③写入Excel用影刀的内置Excel指令【打开Excel文件】→ 目标文件D:\监控数据\{日期}.xlsx 【写入行】→ 将records中的每条记录写入新行 - 列Alocation位置 - 列Bmetric指标 - 列Cvalue数值 -  - 列Dunit单位 - 列Etimestamp时间戳 【保存并关闭】2.4 超标告警逻辑步骤④条件判断在影刀中用【条件判断】节点遍历每条记录条件value 阈值例如温度35°C → 是执行告警流程 → 否继续下一条步骤⑤发送告警【发送企业微信消息】 → 机器人Webhook地址https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的Key → 消息内容 ⚠️ 设备告警 位置{location} 指标{metric} 当前值{value}{unit} 阈值{threshold} 时间{timestamp}2.5 定时执行影刀中设置定时任务【定时触发器】 → 执行频率每5分钟 → 执行流程MQTT数据采集监控 → 执行时间全天三、有什么坑坑1MQTT Broker断连不会自动重连paho-mqtt默认不会自动重连。如果你的MQTT服务器不稳定加上重连逻辑# 替换 client.connect() 部分clientmqtt.Client()client.on_connecton_connect client.on_messageon_message# 启用自动重连client.reconnect_delay_set(min_delay1,max_delay30)try:client.connect(MQTT_BROKER,MQTT_PORT,60)exceptExceptionase:print(f初始连接失败{e})坑2Excel文件写入冲突temu店群自动化报活动案例如果同一个Excel文件被两个流程同时写入影刀会报「文件被占用」的错误。解决方案每个流程写不同的Sheet页或者用数据库SQLite/MySQL代替Excel天然支持并发写入或者在文件名中加时间戳每次写新文件坑3时间戳格式不统一IoT设备的时间戳格式千奇百怪——有的用Unix时间戳秒有的用毫秒有的用ISO格式字符串。统一处理fromdatetimeimportdatetimedefnormalize_timestamp(ts):统一转为 YYYY-MM-DD HH:MM:SS 格式ifisinstance(ts,(int,float)):# 判断是秒还是毫秒ifts1e12:# 毫秒级tsts/1000returndatetime.fromtimestamp(ts).strftime(%Y-%m-%d %H:%M:%S)elifisinstance(ts,str):# 尝试多种格式解析formats[%Y-%m-%dT%H:%M:%S,%Y-%m-%d %H:%M:%S,%Y/%m/%d %H:%M:%S,%d-%m-%Y %H:%M:%S,]forfmtinformats:try:returndatetime.strptime(ts,fmt).strftime(%Y-%m-%d %H:%M:%S)except:continuereturnstr(ts)# 无法解析则返回原值坑4消息积压导致影刀内存溢出如果MQTT Broker上积压了大量消息比如断电后设备恢复一下子推了几千条影刀的Python节点可能内存溢出。预防措施限制单次接收消息数量MAX_MESSAGES100# 单次最多处理100条defon_message(client,userdata,msg):iflen(received_messages)MAX_MESSAGES:return# 达到上限丢弃后续消息# ... 正常处理坑5影刀的Python节点环境问题paho-mqtt库在某些影刀版本中无法直接pip install。如果你遇到ModuleNotFoundError方案A找到影刀内置Python的路径手动安装影刀安装目录\python\python.exe -m pip install paho-mqtt方案B如果实在装不了换用socket原生实现不需要第三方库但代码会复杂很多不推荐新手这样做。总结MQTT 影刀RPA的组合本质上是把物联网协议的实时数据流接入到RPA的定时工作流中。关键在于处理好连接稳定性、时间戳统一、以及Excel/数据库的并发写入问题。配置好之后你只需要在钉钉/企微上等告警消息就行。