关键词：CustomTkinter · Trio · Modbus TCP · 结构化并发 · 共享状态层 · 上位机
一、为什么把 Modbus 轮询托管给 Trio
工业上位机的经典形态是「主线程驱动 GUI、独立线程执行采集」：基于 tkinter/CustomTkinter 渲染界面，开启一个 threading.Thread 循环读取保持寄存器，再通过 queue.Queue 将样本回传主线程刷新控件。
该范式可运行，但在生产环境下存在三处结构性缺陷：
- 生命周期不可控。线程一旦进入 while True，只能依赖 threading.Event 等标志位轮询退出，取消路径容易被遗漏，进而残留不可回收的僵尸线程。
- 故障域不隔离。任一从站掉线或单次读取超时，异常常在线程体内被静默吞噬，表现层呈现「界面卡死」假象，而底层采集协程早已崩溃。
- 并发编排脆弱。多从站、异构轮询周期、存在依赖关系的任务，用线程加锁手工拼装，缺乏统一的作用域管理，可维护性随时间急剧劣化。
Trio 的结构化并发（structured concurrency）模型恰好针对上述问题：所有并发单元都存活于 nursery 作用域内，作用域退出时保证其内部任务被确定性地全部回收；任一任务抛出未捕获异常，整个作用域可通过取消作用域（cancel scope）被统一撤销。本文以一个最小可运行示例，演示如何将 Modbus 轮询真正纳入 Trio 的事件循环，并通过一层共享状态层（shared-state / 内存映射）实现采集与表现的彻底解耦。
二、核心架构：以共享状态层桥接采集与表现
采集单元与表现单元之间最危险的是直接耦合——采集线程直接改写 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
SLAVE = 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, slave=SLAVE)
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 显式让权。
- 异地采集方考虑代理出口（如亿牛云），局域网场景维持直连以最小化时延。
依此骨架落地，上位机界面即可系统性规避「假死」与「僵尸线程」；后续仅需将寄存器点表、缩放系数与报警策略填充进对应抽象即可。
