把 Modbus 轮询塞进 Trio 的异步循环：内存映射与定时采集实战

关键词：CustomTkinter · Trio · Modbus TCP · 结构化并发 · 共享状态层 · 上位机

一、为什么把 Modbus 轮询托管给 Trio

工业上位机常见的做法是：主线程驱动 GUI，独立线程负责采集。具体来说，用 tkinter 或 CustomTkinter 渲染界面，再开一个 threading.Thread 循环读取保持寄存器，最后通过 queue.Queue 把数据传回主线程刷新控件。这套模式确实能跑，但放到生产环境里，存在三个结构性的硬伤。

• 生命周期不可控
  。线程一旦进了 while True，退出就只能靠 threading.Event 这类标志位轮询，取消路径稍微没写对，线程就残留成僵尸，再也收不回来。

• 故障域不隔离
  。任何一个从站掉线，或者单次读取超时，异常往往在线程内部被静默吞掉，表现层看到的却是“界面卡死”，而底层采集协程其实早就崩了。

• 并发编排脆弱
  。多从站、不同轮询周期、任务之间有依赖关系，全靠线程加锁手工拼装，没有统一的作用域管理，维护难度随时间指数级上升。

这时候再看 Trio 的结构化并发模型，会发现它正好对症下药。所有并发单元都存活在 nursery 作用域内，作用域退出时，内部任务会被确定性地全部回收；任何一个任务抛出未捕获异常，整个作用域可以通过取消作用域（cancel scope）统一撤销。下面就用一个最小可运行示例，演示如何把 Modbus 轮询真正纳入 Trio 的事件循环，并且通过一层共享状态层（内存映射）实现采集与表现的彻底解耦。

二、核心架构：以共享状态层桥接采集与表现

采集单元和表现单元之间最怕的就是直接耦合——采集线程直接改 UI 控件的变量，界面一改，协议逻辑也得跟着动。正确的做法是在中间加一层共享状态层（in-memory data model）。它是一个受互斥锁保护的领域对象：采集侧只负责写入，表现侧只负责读取，两边互不感知，依赖关系被这层彻底切断。

同进程内，完全不需引入 multiprocessing.shared_memory，一个受 trio.Lock 保护的 dataclass 就足够了：

import time
from dataclasses import dataclass
from typing import Optional
import trio

@dataclass
class SensorState:
    temperature: float = 0.0
    pressure: float = 0.0
    online: bool = False
    updated_at: float = 0.0

class MemoryMap:
    """采集层与表现层之间的共享状态层，所有访问均受 trio.Lock 串行化。"""

    def __init__(self) -> None:
        self._state = SensorState()
        self._lock = trio.Lock()

    async def update(self, **kwargs) -> None:
        async with self._lock:
            for key, value in kwargs.items():
                setattr(self._state, key, value)
            self._state.updated_at = time.time()

    async def snapshot(self) -> SensorState:
        async with self._lock:
            return SensorState(
                temperature=self._state.temperature,
                pressure=self._state.pressure,
                online=self._state.online,
                updated_at=self._state.updated_at,
            )

snapshot() 返回的是值的副本，而不是原对象的引用。这么做的目的是隔离读写时序：避免表现侧在读的过程中，采集侧并发写入造成部分更新（torn read）。代价只是一次浅拷贝，在典型工业采集频率（100 ms 到 1 s 量级）下，这点开销可以忽略不计。这个模式可以看作单写多读场景下的 Copy-on-Read 策略。

三、将阻塞式 pymodbus 调用桥接进 Trio

pymodbus 的异步客户端是基于 asyncio 事件循环的，而 Trio 用的是独立的内核调度模型，两者没法直接混用。现实的工程选择是：用同步版的 ModbusTcpClient，把它封装进 trio.to_thread.run_sync 派发到默认线程池执行，由 Trio 负责调度和取消语义。

from pymodbus.client import ModbusTcpClient

HOST = "192.168.1.50"
PORT = 502
SLA VE = 1
POLL_INTERVAL = 1.0  # 秒

def _blocking_read() -> Optional[list]:
    """在隔离线程中执行阻塞式 Modbus 读取（功能码 0x03）。"""
    client = ModbusTcpClient(host=HOST, port=PORT, timeout=2.0)
    try:
        if not client.connect():
            return None
        rr = client.read_holding_registers(address=0, count=2, sla ve=SLA VE)
        if rr.isError():
            return None
        return rr.registers  # 两寄存器拼装为 32 位浮点
    finally:
        client.close()

def registers_to_float(regs: list) -> float:
    """按大端字序（ABCD）拼装为 IEEE 754 单精度浮点。"""
    import struct
    return struct.unpack(">f", struct.pack(">HH", regs[0], regs[1]))[0]

async def poll_once(mem: MemoryMap) -> None:
    regs = await trio.to_thread.run_sync(_blocking_read, cancellable=True)
    if regs is None:
        await mem.update(online=False)
        return
    temp = registers_to_float(regs)
    await mem.update(temperature=temp, online=True)

这里有个关键点：cancellable=True 让 Trio 在任务被取消时能够中断底层线程的阻塞式 socket 调用（通过线程级取消机制），避免连接卡死长期占用线程池容量。另外需要明确区分：read_holding_registers（功能码 0x03）对应保持寄存器，read_input_registers（功能码 0x04）对应输入寄存器，量产部署时一定要以从站点表（register map）为准，避免功能码用错。

四、定时采集循环与取消作用域

把采集纳入异步循环，核心在于周期性驱动和超时约束。我们用 trio.move_on_after 给单次轮询套上取消作用域，再用 trio.sleep 锚定轮询周期：

async def poll_loop(mem: MemoryMap) -> None:
    while True:
        with trio.move_on_after(POLL_INTERVAL * 5):
            await poll_once(mem)
        await trio.sleep(POLL_INTERVAL)

然后将 poll_loop 和表现任务一起提交到 nursery，两者就处于同一个结构化并发作用域：

async def main() -> None:
    mem = MemoryMap()
    async with trio.open_nursery() as nursery:
        nursery.start_soon(poll_loop, mem)
        nursery.start_soon(ui_loop, mem)  # CustomTkinter 刷新任务

    # 作用域退出时，所有任务已被确定性回收

这就是结构化并发的核心收益：不用手动 join 线程，也不用清理标志位，nursery 退出就保证所有子任务终结，取消语义由运行时统一保证。

五、与 CustomTkinter 对接

CustomTkinter 的 CTk 实例必须在主线程驱动其事件循环（app.mainloop() 或 app.update()）。表现任务周期性地从 MemoryMap 取快照刷新控件，而采集路径完全不碰任何 UI 对象：

import customtkinter as ctk

async def ui_loop(mem: MemoryMap) -> None:
    app = ctk.CTk()
    label_temp = ctk.CTkLabel(app, text="-- °C")
    label_temp.pack(padx=40, pady=20)
    label_status = ctk.CTkLabel(app, text="连接中…")
    label_status.pack()

    while True:
        snap = await mem.snapshot()
        label_temp.configure(text=f"{snap.temperature:.2f} °C")
        label_status.configure(
            text="在线" if snap.online else "离线",
            text_color="green" if snap.online else "red",
        )
        app.update()  # 非阻塞推进一帧
        await trio.sleep(0.2)  # 让出控制权予采集任务

注意：mainloop() 是阻塞调用，会独占所属线程。所以表现任务必须改用 app.update() 手动推进单帧，并用 await trio.sleep 把控制权交还给调度器；否则采集任务永远得不到执行机会，界面会静止不动。这个模式本质上是“单线程协同多任务”，依赖显式的让权点来维持响应性。

六、异地采集场景：以袋里出口保障链路可用性

前面的示例默认从站位于局域网。但在工程现场，PLC 或边缘网关常常部署在客户侧，而采集程序运行在云端或异地节点，链路要跨公网、NAT 和运营商限速策略。这种远程采集场景下，直连 TCP 会面临三类可用性风险：出口 IP 被对端防火墙限流、跨运营商链路抖动导致丢包、以及多地并发采集时单一源 IP 触及频率阈值。

这时候引入袋里出口就成了合理的工程权衡。袋里服务提供企业级的 IP 资源池，支持 HTTP/HTTPS/SOCKS5 等协议，可以对出站连接做统一 IP 调度。把 Modbus 网关的上行流量经过袋里隧道转发，主要收益包括：

• 链路韧性增强
  ：袋里节点通常具备多线 BGP 接入，相比单条家用宽带，到异地网关的 TCP 握手和长连接保持更稳健，能降低因底层抖动诱发的 ModbusTcpClient 反复重连。

• 源 IP 可治理
  ：当需要多地域并行采集，或者对端按源 IP 做白名单/限速时，袋里服务支持绑定固定出口 IP 或轮换 IP 池，避免单 IP 被打满。

落地层面，同步版 ModbusTcpClient 可以通过自定义 sock 参数接入 SOCKS5 袋里（例如借助 PySocks 预先建立袋里套接字，再交给 pymodbus 复用）。需要明确的是：纯局域网采集完全不需要引入袋里，额外一跳只会增大时延；只有当目标确实在公网且链路可用性不足时，才应该考虑袋里服务作为兜底。选型前建议对袋里节点的时延和丢包率做压测，然后据此决定各从站的袋里/直连路由策略。

七、稳定性工程要点

把 Modbus 轮询托管给 Trio，本质上不是为了“异步而异步”，而是借结构化并发换取三项确定性保障：可取消、可编排、异常可控。再加上共享状态层，采集与表现实现了关注点分离，界面重构不再牵连协议逻辑。落地时需要遵循以下原则：

• 阻塞式 pymodbus 调用必须通过 trio.to_thread.run_sync(..., cancellable=True)，防止线程池被卡死连接耗尽。
• 每次轮询要用 trio.move_on_after 约束超时上界，杜绝单次读取无限挂起。
• 共享状态层的读写统一走 trio.Lock，对外提供不可变快照副本，隔离读写时序。
• CustomTkinter 以 app.update() 手动推进帧，并借 await trio.sleep 显式让权。
• 异地采集考虑袋里出口，局域网场景维持直连以最小化时延。

按照这个骨架落地，上位机界面就能系统性规避“假死”和“僵尸线程”；后续只需要把寄存器点表、缩放系数和报警策略填充进对应的抽象层即可。