影刀RPA MQTT物联网自动化：传感器数据采集与设备监控

影刀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步：

2.3 核心Python代码

步骤①：连接MQTT并接收消息

在影刀中拖入【执行Python代码】节点，写入以下代码：

importpaho.mqtt.clientasmqttimportjsonimporttime# ===== 配置区域（根据实际情况修改）=====MQTT_BROKER="192.168.1.100"# MQTT服务器地址MQTT_PORT=1883# 端口，默认1883MQTT_TOPICS=[("factory/workshop1/temperature",0),# 车间1温度("factory/workshop1/humidity",0),# 车间1湿度("factory/workshop2/temperature",0),# 车间2温度]TIMEOUT=30# 等待消息的超时秒数# ===== 接收消息 =====received_messages=[]defon_connect(client,userdata,flags,rc):"""连接成功的回调"""ifrc==0:print("MQTT连接成功")fortopic,qosinMQTT_TOPICS:client.subscribe(topic,qos)else:print(f"连接失败，返回码：{rc}")defon_message(client,userdata,msg):"""收到消息的回调"""try:payload=json.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}")# 创建客户端并连接client=mqtt.Client()client.on_connect=on_connect client.on_message=on_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字符串，供影刀后续节点使用result=json.dumps(received_messages,ensure_ascii=False)print(f"共收到{len(received_messages)}条消息")

踩坑提醒：time.sleep(TIMEOUT)的时间要合理设置。太短收不到足够的数据，太长流程会卡住。建议先在影刀外部用MQTT客户端测试消息频率，再确定超时时间。

代码执行后，result变量会传给影刀，你需要在影刀中用一个变量（比如mqtt_data）来接收。

步骤②：解析JSON提取数据

再拖入一个【执行Python代码】节点：

importjson# 从上一个节点传入的JSON字符串raw_data=mqtt_data# 影刀变量会自动注入messages=json.loads(raw_data)# 整理成结构化数据records=[]formsginmessages:# 从topic中提取位置信息topic_parts=msg['topic'].split('/')location=topic_parts[1]iflen(topic_parts)>1else'unknown'metric=topic_parts[2]iflen(topic_parts)>2else'unknown'records.append({'location':location,'metric':metric,'value':msg['value'],'unit':msg['unit'],'device_id':msg['device_id'],'timestamp':msg['timestamp'],})# 转为影刀能处理的格式——返回列表的字符串表示# 影刀可以直接用【遍历列表】指令处理output=json.dumps(records,ensure_ascii=False)

步骤③：写入Excel

用影刀的内置Excel指令：

【打开Excel文件】→ 目标文件：D:\监控数据\{日期}.xlsx 【写入行】→ 将records中的每条记录写入新行 - 列A：location（位置） - 列B：metric（指标） - 列C：value（数值） - ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/2ae0a5a7e3c443e49620ab915399fb60.png#pic_center) - 列D：unit（单位） - 列E：timestamp（时间戳） 【保存并关闭】

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数据采集监控 → 执行时间：全天

三、有什么坑

坑1：MQTT Broker断连不会自动重连

paho-mqtt默认不会自动重连。如果你的MQTT服务器不稳定，加上重连逻辑：

# 替换 client.connect() 部分client=mqtt.Client()client.on_connect=on_connect client.on_message=on_message# 启用自动重连client.reconnect_delay_set(min_delay=1,max_delay=30)try:client.connect(MQTT_BROKER,MQTT_PORT,60)exceptExceptionase:print(f"初始连接失败：{e}")

坑2：Excel文件写入冲突

如果同一个Excel文件被两个流程同时写入，影刀会报「文件被占用」的错误。

解决方案：

• 每个流程写不同的Sheet页
• 或者用数据库（SQLite/MySQL）代替Excel，天然支持并发写入
• 或者在文件名中加时间戳，每次写新文件

坑3：时间戳格式不统一

IoT设备的时间戳格式千奇百怪——有的用Unix时间戳（秒），有的用毫秒，有的用ISO格式字符串。

统一处理：

fromdatetimeimportdatetimedefnormalize_timestamp(ts):"""统一转为 'YYYY-MM-DD HH:MM:SS' 格式"""ifisinstance(ts,(int,float)):# 判断是秒还是毫秒ifts>1e12:# 毫秒级ts=ts/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_MESSAGES=100# 单次最多处理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/数据库的并发写入问题。配置好之后，你只需要在钉钉/企微上等告警消息就行。