前言

最近在搞一个工业物联网的小项目：用映翰通 EC942 边缘计算机采集现场设备的数据，再通过 MQTT 把数据传到云端。EC942 自带了 DeviceSupervisor Agent（DSA），配置化点几下鼠标就能采集上云，但我这次想要更灵活的协议处理和数据加工能力，所以选择了自己写采集程序这条路——用 Python 实现一个 Modbus TCP 主站，采集完再用paho-mqtt发布到云端 Broker。

手头没有真实 PLC 怎么办？用Modbus Slave 模拟器在 PC 上模拟一个 Modbus TCP 从站，通过网线直连 EC942，先把整套流程跑通，之后再接真实设备。

今天上午从环境搭建到最终跑通，中间踩了好几个坑，每一个都挺有代表性，整理出来希望能帮后面遇到同样问题的人少走弯路。先上一张整体架构图，看清楚今天到底在打通哪几段链路：

实线是"采集上云"的主链路，虚线是后面加的"云端反向控制"链路。下面按今天实际操作的顺序，一步步还原整个过程。

环境准备

EC942 跑的是 Debian 10 系统，默认 ETH2 网口地址是192.168.4.100/24。我把 PC 网卡配成同网段的静态 IP192.168.4.11，网线直连 EC942 的 ETH2 口——现代网卡都支持 Auto-MDIX，直通线交叉线都能用，不需要交换机。

PC 上打开 Modbus Slave 模拟器，切到 **Modbus TCP Server（从站/服务端）**模式，监听默认的 502 端口，然后在保持寄存器40001~40010里填了一组测试值：1,2,3,4,5,6,7,77,8,9。中间故意插了个77当"特征值"，后面验证解析对不对的时候特别好用——一眼就能看出来有没有对上。

这里有个坑提前说一下：Modbus 协议里的寄存器地址是从0开始的，但模拟器界面显示的是4xxxx这种"传统显示地址"，两者相差1。也就是说界面上的40001对应协议报文里的addr=0，40008（也就是我们的特征值77）对应addr=7。这个"差1"的换算贯穿了后面写代码的整个过程，一定要记住。

踩坑①：ping通了，端口却连不上

万事俱备，先验证一下网络通不通。PC 上ping 192.168.4.100没问题，那反过来在 EC942 上验证到 PC 的连通性总该没问题吧？结果一顿骚操作：

nc -zv 192.168.4.11 502

bash: nc: command not found

换 telnet：

bash: telnet: command not found

那装个 mbpoll 总行了吧：

sudo apt install mbpoll

E: Unable to locate package mbpoll

三连败。原因其实很简单：EC942 跑的是精简版 Debian 镜像，压根没预装这些网络调试工具；而apt install报"找不到包"，是因为Debian 10 (buster) 已经 EOL（生命周期结束），默认的软件源早就下线了，装什么都得先把/etc/apt/sources.list改成指向archive.debian.org才行。

这里还有个概念要理清楚：ping通只能证明网络三层（ICMP）是通的，完全不能说明 502 端口有没有程序在监听。ping 和端口连通性根本是两回事，别被"ping都通了怎么还连不上"这种直觉误导。

工具不好使不代表没法测。EC942 上现成有 Python3，标准库自带的socket模块就能干这个活，不用装任何东西：

python3 -c " import socket try: socket.create_connection(('192.168.4.11', 502), timeout=3) print('端口可达') except OSError as e: print('连接失败:', e) "

这一招后面帮了大忙，记下来备用。

动手写 Modbus TCP 采集代码

工具的坑先放一放，开始写正经代码。Modbus TCP 相比 RTU（串口版本）少了 CRC16 校验（TCP 自身保证可靠传输），但多了一个MBAP 头：

事务标识符(2字节) + 协议标识符(2字节,恒为0) + 长度(2字节) + 单元标识符(1字节) + PDU(功能码+数据)

核心是一个TcpMaster类，内部维护长连接、断线自动重连：

class TcpMaster: def __init__(self, host, port=502, timeout=2.0): self.host = host self.port = port self.timeout = timeout self._sock = None self._txn_id = 0 self._lock = threading.Lock() def _transact(self, unit_id: int, pdu: bytes) -> bytes: with self._lock: try: self._ensure_connected() self._txn_id = (self._txn_id + 1) & 0xFFFF mbap = struct.pack(">HHHB", self._txn_id, 0, len(pdu) + 1, unit_id) self._sock.sendall(mbap + pdu) header = self._recv_exact(7) txn_id, _proto_id, length, _unit_id = struct.unpack( ">HHHB", header) if txn_id != self._txn_id: raise ModbusError(f"事务号不匹配") body = self._recv_exact(length - 1) except (OSError, socket.timeout) as exc: self.close() # 出错即断开，下次自动重连 raise ModbusError(f"TCP 通信异常: {exc}") from exc if body[0] & 0x80: raise ModbusError(f"从站异常码 {body[1] if len(body) > 1 else '未知'}") return body def read_registers(self, unit_id, fc, addr, qty): """fc=3 保持寄存器, fc=4 输入寄存器""" pdu = struct.pack(">BHH", fc, addr, qty) body = self._transact(unit_id, pdu) count = body[1] return list(struct.unpack(f">{count // 2}H", body[2:2 + count]))

点表配置化，不把地址硬编码在代码里，对应上面提到的"40001→addr=0"换算规则：

"points": [ {"name": "reg_40001", "unit_id": 1, "fc": 3, "addr": 0, "qty": 1, "dtype": "uint16"}, {"name": "reg_40008", "unit_id": 1, "fc": 3, "addr": 7, "qty": 1, "dtype": "uint16"} ]

主循环里读点表、拼 JSON、发 MQTT，逻辑很直白：

def poll_once(master, points): values, errors = {}, {} for pt in points: try: regs = master.read_registers(pt["unit_id"], pt["fc"], pt["addr"], pt["qty"]) values[pt["name"]] = round(decode(regs, pt["dtype"], pt.get("scale", 1.0)), 4) except ModbusError as exc: errors[pt["name"]] = str(exc) return values, errors

代码写完了，装好依赖，准备跑起来。

踩坑②：paho-mqtt和Python 3.7八字不合

pip install paho-mqtt python3 main.py config.json

一运行就崩：

File ".../paho/mqtt/client.py", line 49, in <module> from typing import Literal ImportError: cannot import name 'Literal' from 'typing' During handling of the above exception, another exception occurred: ... ModuleNotFoundError: No module named 'typing_extensions'

排查下来发现：EC942 的 Debian 10 自带Python 3.7，而pip install paho-mqtt默认装的是最新的2.1.0。paho-mqtt 从 2.x 开始用到了typing.Literal，这是 Python 3.8 才有的特性，Python 3.7 下会尝试从typing_extensions这个补充包里兜底导入，但环境里压根没装这个包，两条路都走不通，直接报错崩溃。

而且就算装上typing_extensions解决了这个报错，2.x 版本还改了mqtt.Client()的构造函数签名，要求显式传入callback_api_version参数，不传就直接报错——相当于换个坑接着踩。所以最省事的办法是直接把版本锁定到兼容 Python 3.7 的 1.6.1，API 跟老代码完全匹配，不用改一行业务代码：

pip uninstall paho-mqtt -y pip install "paho-mqtt==1.6.1"

这个坑的教训是：装依赖库不能无脑装最新版，尤其是在这种系统 EOL、Python 版本偏老的嵌入式设备上，遇到诡异的 ImportError 先看看是不是库和运行环境的版本对不上。

踩坑③：一个数字引发的连接失败

依赖装好了，重新运行，这次导入没问题了，但是：

WARNING 采集点 reg_40001 失败: TCP 通信异常: [Errno 111] Connection refused WARNING 采集点 reg_40002 失败: TCP 通信异常: [Errno 111] Connection refused ...（十个点位全部失败，每5秒循环一次）

Connection refused（Errno 111）这个报错的含义很明确：TCP 握手包已经送到了目标主机，但对方在那个端口上没有程序监听，直接给拒绝了。这跟网络通不通是两码事——网络层没问题，问题出在"包送去的地方不对，或者对方没在听"。

排查思路很简单，先怀疑配置，再怀疑对端：

1. 检查config.json里tcp.host填的什么 —— 一看，填的是192.168.4.1，而 PC 实际的 IP 是192.168.4.11，少打了最后一个1。
2. 改成正确的192.168.4.11之后重新跑。

一个数字的差别，让程序执着地往一个根本不存在（或者存在但没监听 502 端口）的地址上死磕，报错信息其实已经把线索摆得明明白白，只是当时没往这个方向想。这提醒我一个通用排查顺序：遇到 Connection refused，先把配置文件里的 IP/端口原样打印出来跟实际环境核对一遍，比怀疑代码逻辑更快定位问题。

曙光初现：采集成功

改完 IP，重新运行：

INFO 上报 {'reg_40001': 1.0, 'reg_40002': 2.0, 'reg_40003': 3.0, 'reg_40004': 4.0, 'reg_40005': 5.0, 'reg_40006': 6.0, 'reg_40007': 7.0, 'reg_40008': 77.0, 'reg_40009': 8.0, 'reg_40010': 9.0} rc=0

和模拟器里填的1,2,3,4,5,6,7,77,8,9完全对上，尤其是reg_40008: 77.0这个特征值精确匹配——这就是前面特意埋一个"跳出规律的值"的用处，一眼就能确认地址映射和字节解析全部正确，不用逐个数值去比对。

数值显示成77.0而不是77是正常的，不是bug：解码函数里scale默认值是浮点数1.0，整数乘以浮点数结果自然变成浮点型，不影响数据本身的正确性。

踩坑④：MQTT订阅看不到数据

Modbus 这条链路验证完了，接下来验证数据是不是真的传到了云端 MQTT Broker。用 MQTT.fx 连上 broker，订阅框里填了设备的 topic 前缀factory/line1/ec942-0001，点了 Subscribe：

0 msg/s No content in table

设备端日志明明显示rc=0（发布调用成功），订阅端却啥也收不到，第一反应是不是broker连错了、账号密码不对。仔细一看订阅框里填的主题是：

factory/line1/ec942-0001

而程序实际发布的两条消息在它的子主题下：

factory/line1/ec942-0001/status factory/line1/ec942-0001/telemetry

MQTT 的主题匹配是精确匹配，订阅父级主题不会自动收到子主题的消息，除非用通配符。把订阅改成：

factory/line1/ec942-0001/#

#是多级通配符，能匹配这个前缀下所有层级的子主题。改完立刻就收到了telemetry和status（带online: true的保留消息）两条数据，跟设备端日志的内容完全一致。

这个坑顺带纠正了我对rc=0的一个误解：client.publish()返回的rc=0只表示消息成功放进了本地发送队列，不代表消息已经真正送达远端 Broker。要确认真正送达，必须站在接收方的角度去订阅验证，不能只看发送方日志"没报错"就认为万事大吉。

进阶：让云端也能反向控制设备

采集上云跑通之后，又加了一个功能：云端通过 MQTT 下发指令，反向控制 Modbus 从站的寄存器/线圈。这意味着程序要从单纯的"采集→上报"升级成双向通信：订阅一个命令主题，收到指令后用 Modbus 的写功能码下发，再回一条执行结果的 ack 消息。

先给TcpMaster加上写方法（FC05写线圈、FC06写单寄存器、FC16写多寄存器），复用已有的加锁_transact，读写操作天然线程安全：

def write_register(self, unit_id, addr, value): """fc=6 写单个保持寄存器""" pdu = struct.pack(">BHH", 6, addr, value & 0xFFFF) self._transact(unit_id, pdu)

然后是命令处理回调，收到{"point": "reg_40001", "value": 100}这样的 JSON，查点表、按功能码分发、执行完回 ack：

def make_command_handler(master, control_index, ack_topic): def _on_message(client, _userdata, msg): point_name = None try: cmd = json.loads(msg.payload.decode("utf-8")) point_name = cmd["point"] value = cmd["value"] pt = control_index[point_name] # 白名单：不在点表里的点位名直接 KeyError fc = pt["fc"] if fc == 6: master.write_register(pt["unit_id"], pt["addr"], int(value)) # ... fc==5写线圈、fc==16写多寄存器同理 ack = {"point": point_name, "value": value, "success": True} except (KeyError, ValueError, TypeError, ModbusError) as exc: ack = {"point": point_name, "success": False, "error": str(exc)} client.publish(ack_topic, json.dumps(ack), qos=1) return _on_message

测试的时候直接复用了现有环境：用 MQTT.fx 往factory/line1/ec942-0001/cmd发布{"point": "reg_40001", "value": 100}，PC 上 Modbus Slave 窗口里40001立刻变成了100，下一轮采集的telemetry里也同步更新——写入和读回形成了完整闭环。

这里必须多说一句安全上的考虑：写操作比读操作危险得多，读错了顶多是数据不对，写错了是真的会影响现场设备。所以代码里做了两层防护：一是点位白名单，control_index只认control_points里显式列出的点位，其他点位名直接被拒绝；二是所有写操作不管成功失败都会记日志，方便事后审计。生产环境上线前，MQTT Broker 那边的账号权限也得单独收紧，别让所有客户端账号都能往控制指令主题发消息。

总结：今天的踩坑清单

从踩坑的分布能看出来，今天遇到的坑大多不是"协议本身难"，而是环境细节：一个精简系统缺个工具、一个库版本不兼容、一个IP多打少打一位、一个MQTT主题没加通配符。这类问题排查起来比写代码逻辑本身更磨人，也更容易被忽略，希望这篇记录能帮到同样在踩坑路上的朋友。

完整代码结构：

├── config.json # 点表配置（读点位 + 写点位）+ MQTT连接参数 ├── modbus_tcp.py # Modbus TCP 主站实现（读写寄存器/线圈） └── main.py # 主循环：采集轮询 + MQTT发布 + 控制指令订阅

如果这篇文章对你有帮助，欢迎点赞收藏，后续如果接入真实PLC或者对接私有协议，我会继续更新踩坑记录