pymodbus与Modbus TCP集成：完整示例说明

用 Python 打通工业现场：pymodbus + Modbus TCP 实战全解析

你有没有遇到过这样的场景？产线上的 PLC 只支持 Modbus 协议，而你的数据分析平台是用 Python 写的；你想做个实时监控页面，却发现组态软件定制成本太高、改起来太慢。这时候，pymodbus就是你最该掌握的“翻译官”。

在工业自动化领域，Modbus 是那个“老而弥坚”的通信协议——它简单、开放、兼容性极强，至今仍是连接 PLC、传感器、变频器等设备的事实标准。而在现代开发中，Python 凭借其强大的生态和快速迭代能力，成为边缘计算、数据采集与系统集成的首选语言。

那么问题来了：如何让这两个世界无缝对接？

答案就是pymodbus——一个纯 Python 实现的 Modbus 协议栈，专为像你我这样的开发者设计。本文不讲空泛理论，而是带你从零开始，一步步搭建一个完整的 Modbus TCP 通信系统：客户端读数据、服务器模拟设备、异步高并发处理、Web 接口封装……所有关键环节一网打尽。

为什么选择 pymodbus？不只是“能用”

市面上其实有不少 Modbus 工具库，比如 C 语言的libmodbus，或者一些闭源 SDK。但如果你正在做的是边缘网关、数据中台或轻量级 SCADA 系统，你会发现这些方案要么太重，要么太难集成。

而pymodbus的优势在于：

纯 Python 编写，无需编译，安装一条命令搞定：pip install pymodbus
支持同步与异步模式，既能写脚本也能扛高并发
完整支持 Modbus TCP、RTU over TCP、串行 RTU/ASCII
提供可扩展的数据存储模型，甚至可以自定义后端（比如数据库）
调试友好，日志清晰，报文可打印，Wireshark 抓包直接对得上

更重要的是，它可以轻松和 Flask、FastAPI、Pandas、MQTT 这些现代工具链打通。这意味着你可以用几行代码就把车间里的温度传感器数据变成网页图表，而不是花几周去学某个工控组态软件。

先跑通第一个例子：启动一个 Modbus 服务器

我们先来模拟一个“从站设备”——比如一台支持 Modbus TCP 的温控仪表。它对外暴露几个寄存器，主站可以通过功能码 0x03 读取当前值。

下面是使用 pymodbus 搭建 Modbus TCP 服务器的核心代码：

from pymodbus.server import StartTcpServer from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSlaveContext, ModbusServerContext import logging # 启用详细日志，方便调试 logging.basicConfig(level=logging.DEBUG) def run_server(): # 创建一块保持寄存器区域，地址0起始，初始值 [10, 20, 30, 40] block = ModbusSequentialDataBlock(0x00, [10, 20, 30, 40]) # 构建从站上下文，只初始化 holding registers (hr) store = ModbusSlaveContext(hr=block) context = ModbusServerContext(slaves=store, single=True) print("🚀 启动 Modbus TCP 服务器，监听 502 端口...") StartTcpServer(context, address=("localhost", 502)) if __name__ == "__main__": run_server()

关键点解读：

ModbusSequentialDataBlock(0x00, [10, 20, 30, 40])：创建一段连续的寄存器块，对应 Modbus 地址 0~3。
hr=表示 holding registers，也就是通常用来读写的 4xxxx 区域（如 40001）。
single=True表示这是一个单设备模式，适用于大多数测试场景。
默认监听502 端口，这是 IANA 为 Modbus TCP 注册的标准端口。

运行这个脚本后，你就拥有了一台“虚拟仪表”，等待别人来读它的数据了。

💡 小技巧：如果你想模拟多个设备（Unit ID 不同），可以把slaves={1: slave1, 2: slave2}传入ModbusServerContext，然后通过 MBAP 头中的 Unit ID 来区分请求目标。

再写一个客户端：主动去读数据

现在我们换角色，做一个“主站”程序，去定时读取刚才那台“仪表”的数据。

from pymodbus.client import ModbusTcpClient import time def run_client(): client = ModbusTcpClient('localhost', port=502) if not client.connect(): print("❌ 连接失败，请检查服务器是否运行") return print("✅ 已连接到 Modbus 服务器") try: while True: # 读取地址0开始的4个保持寄存器 result = client.read_holding_registers(address=0, count=4, slave=1) if result.isError(): print(f"⚠️ Modbus 错误: {result}") else: print(f"📈 读取成功: {result.registers}") time.sleep(2) except KeyboardInterrupt: print("\n👋 客户端已停止") finally: client.close() if __name__ == "__main__": run_client()

输出效果：

✅ 已连接到 Modbus 服务器 📈 读取成功: [10, 20, 30, 40] 📈 读取成功: [10, 20, 30, 40] ...

注意事项：

slave=1对应 MBAP 中的 Unit ID，必须和服务端一致。
每次调用read_holding_registers都是一次独立事务，会生成新的 Transaction ID。
建议加上异常捕获，防止网络中断导致程序崩溃。

想提升性能？试试异步客户端

上面的例子是同步阻塞的：每次读取都要等响应回来才能继续。如果要同时监控几十台设备，效率就很低。

幸运的是，pymodbus 支持基于asyncio的异步模式，让你在一个线程里高效管理多个连接。

from pymodbus.client import AsyncModbusTcpClient import asyncio async def fetch_data(client, name, addr, count): try: result = await client.read_holding_registers(address=addr, count=count, slave=1) if not result.isError(): print(f"{name}: {result.registers}") else: print(f"{name} 读取失败: {result}") except Exception as e: print(f"{name} 异常: {e}") async def run_async_client(): client = AsyncModbusTcpClient('localhost', port=502) await client.connect() print("🔗 异步客户端已连接") try: for i in range(5): # 并发读取（这里只是同一个设备，实际可扩展为多个IP） await fetch_data(client, f"设备{i+1}", 0, 4) await asyncio.sleep(2) finally: client.close() if __name__ == "__main__": asyncio.run(run_async_client())

优势在哪？

使用await而非time.sleep()，释放 CPU 资源；
多个fetch_data可并行发起（配合任务调度如asyncio.gather）；
更适合用于边缘网关轮询多台设备的场景。

Modbus TCP 报文长什么样？抓包一看便知

很多人觉得 Modbus “黑盒”，其实是没看过真实报文。下面我们拆解一次典型的读寄存器请求。

假设客户端发送如下请求：

[00 01] → Transaction ID = 1 [00 00] → Protocol ID = 0 [00 06] → Length = 6 字节（后续内容） [01] → Unit ID = 1 [03] → Function Code = 0x03（读保持寄存器） [00 00] → 起始地址 = 0 [00 04] → 寄存器数量 = 4

总长度 12 字节，结构为：MBAP(7) + PDU(5)。

服务端返回：

[00 01][00 00][00 05][01][03][08][00 0A 00 14 00 1E 00 28]

其中[08]是字节数（8个字节数据），后面是 4 个寄存器的值：10, 20, 30, 40（十六进制表示）。

✅ 提示：用 Wireshark 抓包时，搜索tcp.port == 502即可看到清晰解码后的 Modbus 请求/响应。

实际部署要考虑什么？这些坑我都踩过

别以为跑通 demo 就万事大吉。真正把 pymodbus 用到生产环境，还得注意以下几点：

1. 自动重连机制不能少

网络波动太常见了。建议封装一个带重试逻辑的客户端：

def connect_with_retry(client, max_retries=5): for i in range(max_retries): if client.connect(): print("✅ 连接成功") return True print(f"🔁 第 {i+1} 次重连失败，2秒后重试...") time.sleep(2) return False

2. 避免多线程/协程争抢同一个 client

ModbusTcpClient不是线程安全的！如果你在多个线程或协程中共享同一个实例，可能会出现事务 ID 冲突或响应错乱。

✅ 正确做法：每个任务单独创建 client，或加锁控制访问顺序。

3. 合理设置超时时间

默认超时可能是 3 秒，但在复杂网络下可能不够：

client = ModbusTcpClient('192.168.1.100', port=502, timeout=5.0)

也可以为每个操作单独指定：

result = client.read_holding_registers(0, 4, timeout=3)

4. 功能码别搞混！

功能码	名称	数据区
0x01	读线圈状态	Coils (0xxxx)
0x02	读离散输入	Discretes (1xxxx)
0x03	读保持寄存器	HR (4xxxx)
0x04	读输入寄存器	IR (3xxxx)
0x05	写单个线圈
0x06	写单个保持寄存器
0x10	写多个保持寄存器

记住一句话：读数据优先看 0x03 和 0x04；写参数用 0x06 或 0x10。

如何让它接入现代系统？Web API + MQTT 示例

真正的价值不是“读到数据”，而是“让数据流动起来”。下面教你两招实用集成技巧。

方法一：用 Flask 暴露 RESTful 接口

from flask import Flask, jsonify from pymodbus.client import ModbusTcpClient app = Flask(__name__) client = ModbusTcpClient('localhost', port=502) @app.route('/api/sensors') def get_sensors(): if not client.connect(): return jsonify({"error": "连接失败"}), 500 try: result = client.read_holding_registers(address=0, count=4) client.close() if result.isError(): return jsonify({"error": "读取失败"}), 500 data = { "temperature": result.registers[0], "pressure": result.registers[1], "flow_rate": result.registers[2], "level": result.registers[3] } return jsonify(data) except Exception as e: client.close() return jsonify({"error": str(e)}), 500 if __name__ == '__main__': app.run(port=5000)

启动后访问http://localhost:5000/api/sensors，就能拿到 JSON 格式的数据，前端随便画图。

方法二：上传到 MQTT，实现远程监控

import paho.mqtt.client as mqtt from pymodbus.client import ModbusTcpClient import json import time mqtt_client = mqtt.Client() mqtt_client.connect("broker.emqx.io", 1883, 60) modbus_client = ModbusTcpClient('localhost', 502) while True: if modbus_client.connect(): result = modbus_client.read_holding_registers(0, 4) modbus_client.close() if not result.isError(): payload = { "temp": result.registers[0], "press": result.registers[1], "ts": int(time.time()) } mqtt_client.publish("factory/sensor/data", json.dumps(payload)) print("📤 数据已发布至 MQTT") else: print("⚠️ Modbus 连接失败") time.sleep(5)

结合 Grafana + InfluxDB + MQTT 订阅，一套完整的 IIoT 监控系统就成型了。