影刀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代码】节点，写入以下代码：

import paho.mqtt.client as mqtt
import json
import time

# ===== 配置区域（根据实际情况修改）=====
MQTT_BROKER = "192.168.1.100"  # MQTT服务器地址
MQTT_PORT = 1883                # 端口，默认1883
MQTT_TOPICS = [
    ("factory/workshop1/temperature", 0),  # 车间1温度
    ("factory/workshop1/humidity", 0),     # 车间1湿度
    ("factory/workshop2/temperature", 0),  # 车间2温度
]
TIMEOUT = 30  # 等待消息的超时秒数

# ===== 接收消息 =====
received_messages = []

def on_connect(client, userdata, flags, rc):
    """连接成功的回调"""
    if rc == 0:
        print("MQTT连接成功")
        for topic, qos in MQTT_TOPICS:
            client.subscribe(topic, qos)
    else:
        print(f"连接失败，返回码：{rc}")

def on_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', ''),
        })
    except Exception as e:
        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代码】节点：

import json

# 从上一个节点传入的JSON字符串
raw_data = mqtt_data  # 影刀变量会自动注入
messages = json.loads(raw_data)

# 整理成结构化数据
records = []
for msg in messages:
    # 从topic中提取位置信息
    topic_parts = msg['topic'].split('/')
    location = topic_parts[1] if len(topic_parts) > 1 else 'unknown'
    metric = topic_parts[2] if len(topic_parts) > 2 else '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)
except Exception as e:
    print(f"初始连接失败：{e}")

坑2：Excel文件写入冲突

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

解决方案：

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

坑3：时间戳格式不统一

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

统一处理：

from datetime import datetime

def normalize_timestamp(ts):
    """统一转为 'YYYY-MM-DD HH:MM:SS' 格式"""
    if isinstance(ts, (int, float)):
        # 判断是秒还是毫秒
        if ts > 1e12:  # 毫秒级
            ts = ts / 1000
        return datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S')
    elif isinstance(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',
        ]
        for fmt in formats:
            try:
                return datetime.strptime(ts, fmt).strftime('%Y-%m-%d %H:%M:%S')
            except:
                continue
    return str(ts)  # 无法解析则返回原值

坑4：消息积压导致影刀内存溢出

如果MQTT Broker上积压了大量消息（比如断电后设备恢复，一下子推了几千条），影刀的Python节点可能内存溢出。

预防措施：限制单次接收消息数量：

MAX_MESSAGES = 100  # 单次最多处理100条

def on_message(client, userdata, msg):
    if len(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/数据库的并发写入问题。配置好之后，你只需要在钉钉/企微上等告警消息就行。