如何从无限运行的asyncio事件循环外部访问数据对象？

核心思路 #

因为StartAsyncTcpServer启动后会进入无限await状态无法返回实例，所以需要通过共享可变容器或信号传递的方式，在协程创建服务器实例后立即将其保存到外部可访问的位置。同时要注意和PyQt5 GUI的线程兼容问题，避免阻塞主线程。

具体实现方案 #

方案1：使用共享可变容器存储实例 #

适合不需要和GUI整合，或手动处理线程的场景：

1. 定义一个共享的可变对象（比如字典），用于保存服务器实例：

server_storage = {}

2. 拆分服务器创建和启动逻辑，将实例存入共享容器：

async def run_server_simulator(args, storage):
    # 直接创建ModbusTcpServer实例（替代原StartAsyncTcpServer，因为原函数会直接await无限运行）
    from pymodbus.server import AsyncModbusTcpServer
    server = AsyncModbusTcpServer(
        context=args.context,
        address=("", args.port),
        framer=args.framer,
        allow_reuse_address=True,
    )
    # 将实例存入共享容器
    storage['instance'] = server
    # 启动服务器的无限运行
    await server.serve_forever()

def start_server(storage):
    cmd_args = get_commandline()
    run_args = setup_simulator(cmd_args)
    # 运行协程（注意：asyncio.run会阻塞当前线程）
    asyncio.run(run_server_simulator(run_args, storage), debug=True)

3. 顶层访问实例：

# 在子线程启动服务器，避免阻塞主线程
import threading
threading.Thread(target=start_server, args=(server_storage,), daemon=True).start()

# 后续需要时访问实例
if 'instance' in server_storage:
    server = server_storage['instance']
    # 修改寄存器示例：修改保持寄存器地址0的值为456
    server.context.setValues(0x03, 0, [456])

方案2：结合PyQt5的线程与信号传递 #

因为PyQt5的GUI需要主线程处理事件，必须将asyncio服务器放到子线程运行，并通过信号传递实例：

1. 定义承载服务器的QThread子类：

from PyQt5.QtCore import QThread, pyqtSignal
import asyncio
from pymodbus.server import AsyncModbusTcpServer

class ServerThread(QThread):
    # 定义信号，用于传递服务器实例
    server_ready = pyqtSignal(object)

    def __init__(self, run_args):
        super().__init__()
        self.run_args = run_args
        self.server = None

    def run(self):
        # 在子线程创建独立的asyncio事件循环
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        async def start_server():
            # 创建服务器实例
            self.server = AsyncModbusTcpServer(
                context=self.run_args.context,
                address=("", self.run_args.port),
                framer=self.run_args.framer,
                allow_reuse_address=True,
            )
            # 发送信号通知GUI服务器已就绪
            self.server_ready.emit(self.server)
            # 启动无限运行的服务
            await self.server.serve_forever()

        # 运行协程
        loop.run_until_complete(start_server())

2. 在GUI窗口中使用该线程并接收实例：

from PyQt5.QtWidgets import QWidget, QPushButton, QVBoxLayout

class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.init_ui()
        self.init_server()

    def init_ui(self):
        self.layout = QVBoxLayout()
        self.modify_btn = QPushButton("修改保持寄存器值", self)
        self.modify_btn.clicked.connect(self.on_modify_click)
        self.layout.addWidget(self.modify_btn)
        self.setLayout(self.layout)
        self.server = None

    def init_server(self):
        # 初始化服务器参数
        cmd_args = get_commandline()
        run_args = setup_simulator(cmd_args)
        # 创建并启动服务器线程
        self.server_thread = ServerThread(run_args)
        self.server_thread.server_ready.connect(self.on_server_ready)
        self.server_thread.start()

    def on_server_ready(self, server):
        self.server = server
        print("服务器实例已获取")

    def on_modify_click(self):
        if not self.server:
            print("服务器未就绪")
            return
        
        # 跨线程访问asyncio对象需通过事件循环线程安全执行
        loop = asyncio.get_event_loop()
        async def update_register():
            # 修改保持寄存器地址0的值为1234
            self.server.context.setValues(0x03, 0, [1234])
            print("寄存器值已修改")
        
        # 提交协程到服务器的事件循环执行
        future = asyncio.run_coroutine_threadsafe(update_register(), loop)
        try:
            future.result()
        except Exception as e:
            print(f"修改失败: {str(e)}")

关键注意事项 #

• 线程安全：Modbus服务器实例和asyncio事件循环不是线程安全的，所有对服务器的修改操作必须提交到对应的asyncio事件循环中执行，使用asyncio.run_coroutine_threadsafe或loop.call_soon_threadsafe。
• 避免阻塞主线程：asyncio.run会阻塞当前线程，GUI场景下必须将服务器放到子线程运行，否则会导致GUI冻结。
• API适配：如果使用的Modbus库中StartAsyncTcpServer不允许拆分创建和启动步骤，需查看库文档，找到直接创建服务器实例的方法（比如pymodbus中的AsyncModbusTcpServer类）。

内容的提问来源于stack exchange，提问作者jie