如何正确使用qasync实现PyQt异步UDP通信？

问题根源 #

错误码0xC0000409是线程资源冲突或非法调度导致的崩溃，你的代码核心问题有两点：

• 在子线程中错误初始化QEventLoop并调用asyncio.set_event_loop，这会和主线程的事件循环产生严重冲突
• 错误使用QThread搭配qasync，qasync的设计初衷是整合Qt与asyncio的事件循环，不需要额外创建线程跑UDP逻辑

修复后的完整代码 #

import sys
import asyncio
from PyQt5.QtCore import pyqtSignal, QObject, pyqtSlot
from qasync import QEventLoop, asyncSlot
from PyQt5.QtWidgets import QApplication, QMainWindow
from gui import Ui_MainWindow

class UDPProtocol(asyncio.DatagramProtocol):
    def __init__(self, data_emitter):
        self.transport = None
        self.data_emitter = data_emitter

    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        self.data_emitter(data, addr)

    def error_received(self, exc):
        print(f'Error Received: {exc}')

    def connection_lost(self, exc):
        print(f'Connection Closed: {exc}')
        self.transport = None

class UDPClient(QObject):
    data_received = pyqtSignal(bytes, tuple)

    def __init__(self, local_addr):
        super().__init__()
        self.local_addr = local_addr
        self.transport = None
        self.protocol = None
        self._server_task = None

    async def start_server(self):
        loop = asyncio.get_running_loop()
        self.transport, self.protocol = await loop.create_datagram_endpoint(
            lambda: UDPProtocol(self.emit_data),
            local_addr=self.local_addr
        )

    def emit_data(self, data, addr):
        self.data_received.emit(data, addr)

    def send(self, msg, addr):
        if self.transport:
            try:
                print('Sent')
                self.transport.sendto(msg, addr)
            except Exception as e:
                print(f"[SEND ERROR] {e}")
        else:
            print("[SEND ERROR] UDP transport not initialized")

    async def close(self):
        if self.transport:
            self.transport.close()
            await asyncio.sleep(0.1)

class MainWindow(QMainWindow, Ui_MainWindow):
    send_message = pyqtSignal(bytes, tuple)

    def __init__(self):
        super().__init__()
        self.setupUi(self)
        self.udp = UDPClient(('192.168.68.58', 9999))
        self.hp = ('192.168.68.56', 9999)

        # 信号连接
        self.forwardButton.clicked.connect(self.send_fwd)
        self.startButton.clicked.connect(self.on_start_clicked)
        self.udp.data_received.connect(self.handle_received)
        self.send_message.connect(self.udp.send)

    def send_fwd(self):
        msg = b'hello'
        self.send_message.emit(msg, self.hp)

    @asyncSlot()
    async def on_start_clicked(self):
        if not self.udp._server_task:
            self.udp._server_task = asyncio.create_task(self.udp.start_server())
            try:
                await self.udp._server_task
            except Exception as e:
                print(f"[START ERROR] {e}")
                self.udp._server_task = None

    @pyqtSlot(bytes, tuple)
    def handle_received(self, data, addr):
        print(f"[RECV] {data.decode()}")

def main():
    app = QApplication(sys.argv)
    # 整合Qt与asyncio事件循环
    loop = QEventLoop(app)
    asyncio.set_event_loop(loop)

    gui = MainWindow()
    gui.show()

    with loop:
        sys.exit(loop.run_forever())

if __name__ == "__main__":
    main()

关键修复点 #

• 移除多余的QThread，使用qasync.QEventLoop统一管理所有异步任务与Qt界面事件，彻底避免线程冲突
• 将UDPClient的启动逻辑改为异步函数，通过asyncSlot在主线程事件循环中调度，确保异步任务执行环境合法
• 删除UDPClient中自行创建的QEventLoop，直接复用qasync提供的全局事件循环
• 添加UDP传输初始化状态判断，避免空指针调用引发异常

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