当前位置：首页 > news >正文

保姆级教程：用Python监听EMQX设备上下线，并实时写入MySQL数据库

news 2026/4/29 20:09:42

Python实战：EMQX设备状态监控与MySQL持久化全流程解析

物联网项目中，设备连接状态的实时监控是业务系统的基础需求。今天我将分享一个完整的解决方案：用Python监听EMQX的MQTT系统主题，捕获设备上下线事件，并将这些状态变化实时记录到MySQL数据库。这套方案已经在多个工业物联网项目中稳定运行，下面拆解每个技术环节。

1. 环境准备与EMQX配置

在开始编码前，我们需要完成基础环境搭建。不同于简单的Demo环境，生产级部署需要考虑权限控制和系统稳定性。

必备组件清单：

EMQX 5.0+（建议使用企业版）
Python 3.8+（推荐3.10+版本）
MySQL 8.0+（支持JSON字段的版本）
网络环境确保Python服务器能访问EMQX的1883端口

1.1 EMQX ACL配置

默认情况下，EMQX禁止普通客户端订阅$SYS系统主题。我们需要通过Dashboard修改ACL规则：

登录EMQX Dashboard
导航到"访问控制" → "授权"
在File数据源中添加规则：

{allow, {user, "monitor_client"}, subscribe, ["$SYS/brokers/+/clients/#"]}.

这条规则允许用户名为monitor_client的客户端订阅所有设备上下线事件。

生产环境建议使用更精确的IP限制，例如：{allow, {ipaddr, "192.168.1.100"}, subscribe, ["$SYS/brokers/+/clients/#"]}

2. Python监听实现

我们将使用paho-mqtt库实现MQTT客户端。这个库虽然基础，但在处理系统主题时需要特别注意消息回调的线程安全。

2.1 基础连接代码

import paho.mqtt.client as mqtt import json import threading class DeviceMonitor: def __init__(self): self.client = mqtt.Client(client_id="status_monitor_1") self.client.username_pw_set("monitor_client", "your_password") # 绑定回调函数 self.client.on_connect = self.on_connect self.client.on_message = self.on_message def on_connect(self, client, userdata, flags, rc): if rc == 0: print("Connected to EMQX") # 订阅所有设备的上下线事件 client.subscribe("$SYS/brokers/+/clients/+/connected") client.subscribe("$SYS/brokers/+/clients/+/disconnected") else: print(f"Connection failed with code {rc}") def on_message(self, client, userdata, msg): try: payload = json.loads(msg.payload.decode()) print(f"Received message on {msg.topic}: {payload}") # 这里添加后续处理逻辑 self.process_device_event(msg.topic, payload) except Exception as e: print(f"Error processing message: {e}") def start(self): self.client.connect("your_emqx_host", 1883, 60) # 使用单独线程运行网络循环 self.client.loop_start() monitor = DeviceMonitor() monitor.start()

2.2 消息处理优化

原始实现有几个潜在问题：

没有重连机制
消息处理阻塞网络线程
缺乏消息队列缓冲

改进后的处理架构：

from queue import Queue import time class EnhancedDeviceMonitor(DeviceMonitor): def __init__(self): super().__init__() self.event_queue = Queue(maxsize=1000) self.processing_thread = threading.Thread(target=self.process_queue) self.running = True def on_message(self, client, userdata, msg): try: payload = json.loads(msg.payload.decode()) # 非阻塞放入队列 self.event_queue.put_nowait((msg.topic, payload)) except Exception as e: print(f"Error handling message: {e}") def process_queue(self): while self.running: try: topic, payload = self.event_queue.get(timeout=1) self.process_device_event(topic, payload) except Empty: continue def start(self): super().start() self.processing_thread.start() def stop(self): self.running = False self.processing_thread.join() self.client.disconnect()

3. MySQL数据存储设计

设备状态记录需要平衡查询效率和存储空间。以下是经过验证的表结构设计：

3.1 数据库表结构

CREATE TABLE device_status ( id BIGINT AUTO_INCREMENT PRIMARY KEY, device_id VARCHAR(64) NOT NULL, status ENUM('online', 'offline') NOT NULL, event_time TIMESTAMP(3) NOT NULL, client_info JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_device (device_id), INDEX idx_time (event_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

关键设计点：

使用TIMESTAMP(3)存储毫秒级时间戳
JSON字段保存完整的客户端信息
复合索引优化常见查询场景

3.2 Python数据入库实现

推荐使用SQLAlchemy ORM，它提供了连接池管理和事务处理：

from sqlalchemy import create_engine, Column, Integer, String, TIMESTAMP, Enum, JSON from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class DeviceStatus(Base): __tablename__ = 'device_status' id = Column(Integer, primary_key=True) device_id = Column(String(64), nullable=False) status = Column(Enum('online', 'offline'), nullable=False) event_time = Column(TIMESTAMP(precision=3), nullable=False) client_info = Column(JSON) created_at = Column(TIMESTAMP) # 初始化数据库连接 engine = create_engine( "mysql+pymysql://user:password@localhost/iot_db", pool_size=5, max_overflow=10, pool_recycle=3600 ) Session = sessionmaker(bind=engine) class DatabaseWriter: def __init__(self): self.session = Session() def write_status(self, device_id, status, event_time, client_info): try: record = DeviceStatus( device_id=device_id, status=status, event_time=event_time, client_info=client_info ) self.session.add(record) self.session.commit() except Exception as e: self.session.rollback() print(f"Database error: {e}") finally: self.session.close()

4. 完整系统集成

现在我们将各个模块组合成完整解决方案：

4.1 事件处理器实现

from datetime import datetime class EventProcessor: def __init__(self): self.db_writer = DatabaseWriter() def parse_topic(self, topic): """解析主题获取设备ID""" parts = topic.split('/') return parts[-2] # clientid位置 def process_device_event(self, topic, payload): device_id = self.parse_topic(topic) status = 'online' if 'connected' in topic else 'offline' # 转换时间戳 event_ts = payload.get('connected_at') or payload.get('disconnected_at') event_time = datetime.fromtimestamp(event_ts/1000) self.db_writer.write_status( device_id=device_id, status=status, event_time=event_time, client_info=payload )

4.2 系统监控与告警

生产环境还需要添加监控指标：

from prometheus_client import start_http_server, Counter, Gauge # 定义监控指标 EVENTS_PROCESSED = Counter( 'device_events_processed_total', 'Total processed device events', ['status'] ) QUEUE_SIZE = Gauge( 'device_event_queue_size', 'Current size of processing queue' ) class MonitoredEventProcessor(EventProcessor): def process_device_event(self, topic, payload): QUEUE_SIZE.set(self.event_queue.qsize()) try: super().process_device_event(topic, payload) EVENTS_PROCESSED.labels(status='success').inc() except Exception: EVENTS_PROCESSED.labels(status='failed').inc() # 启动监控服务器 start_http_server(8000)

5. 性能优化技巧

在实际部署中，我们总结了这些优化经验：

5.1 批量写入优化

高频设备场景下，单条写入会成为瓶颈。改用批量写入：

from sqlalchemy.dialects.mysql import insert def batch_write(self, events): """批量写入设备事件""" if not events: return records = [{ 'device_id': e['device_id'], 'status': e['status'], 'event_time': e['event_time'], 'client_info': e['client_info'] } for e in events] stmt = insert(DeviceStatus).values(records) do_update_stmt = stmt.on_duplicate_key_update( status=stmt.inserted.status, event_time=stmt.inserted.event_time ) with engine.begin() as conn: conn.execute(do_update_stmt)

5.2 连接管理策略

推荐配置参数：

engine = create_engine( "mysql+pymysql://user:password@localhost/iot_db", pool_size=10, # 常规连接数 max_overflow=20, # 临时增加的连接数 pool_timeout=30, # 获取连接超时时间(秒) pool_recycle=3600, # 连接回收时间(秒) pool_pre_ping=True # 执行前检查连接有效性 )

6. 异常处理与日志

工业级应用需要完善的错误处理：

import logging from logging.handlers import RotatingFileHandler # 配置日志 logging.basicConfig( handlers=[ RotatingFileHandler( 'device_monitor.log', maxBytes=10*1024*1024, # 10MB backupCount=5 ) ], level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) class ResilientEventProcessor(EventProcessor): def process_device_event(self, topic, payload): try: # ...原有处理逻辑... except json.JSONDecodeError: logging.error(f"Invalid JSON payload from {topic}") except sqlalchemy.exc.DBAPIError as db_err: logging.error("Database error", exc_info=db_err) self.reconnect_db() except Exception as e: logging.critical("Unexpected error", exc_info=e) def reconnect_db(self): try: self.session.rollback() self.session = Session() except Exception as e: logging.error("DB reconnection failed", exc_info=e) time.sleep(5) # 避免快速重试

这套系统在智能工厂项目中稳定运行了18个月，日均处理设备事件超过200万条。最关键的经验是：一定要为MQTT客户端实现完善的断线重连机制，并在数据库写入层做好批量处理和错误恢复。

查看全文

http://www.jsqmd.com/news/720817/