如何使用 asyncio.serve_forever() 而不冻结 GUI？

Question

我正在尝试使用 pyqt5 创建 GUI 以进行 tcp 通信。我有 2 个不同的 .py 文件，一个用于 GUI，一个用于 TCP。 tcp.py 包含 tcpClients 和 tcpServer 类，其中有一个名为 tcp_server_connect 的函数（可以在下面找到）来创建客户端和服务器之间的连接。 GUI 有一个名为 connect 的按钮，我想在按下它时调用函数 tcp_server_connect 。当我按下它时，GUI 冻结并且没有响应。没有错误。我认为这是因为 server.serve_forever() 循环，但我不知道如何解决这个问题。

async def tcp_server_connect():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)
    async with server:
        await server.serve_forever()

Answer 1

问题是 Qt 在主线程上运行，尝试在同一线程上运行

asyncio

事件循环将不起作用。 Qt 和事件循环都想要并且需要控制整个线程。请注意，Qt 最近确实宣布了一个用于 Python 的

asyncio

模块，但仍处于技术预览版。

下面的示例使用 PySide6（因为我已经安装了），有一个 Qt 窗口，其中有用于启动 TCP 服务器的按钮，然后是一个字符串输入控件和一个用于将消息发送到 TCP 服务器的按钮。有一个

asyncio

事件循环在一个单独的线程上运行，该线程侦听消息，然后执行所需的操作。试试看。您需要安装的只是 PySide6。

# Core dependencies
import asyncio
import sys
from threading import Thread

# Package dependencies
from PySide6.QtWidgets import (
    QApplication,
    QWidget,
    QPushButton,
    QVBoxLayout,
    QLineEdit,
)


class MainWindow(QWidget):
    def __init__(self) -> None:
        super().__init__()

        # The the `asyncio` queue and event loop are created here, in the GUI thread (main thread),
        # but they will be passed into a new thread that will actually run the event loop.
        # Under no circumstances should the `asyncio.Queue` be used outside of that event loop. It
        # is only okay to construct it outside of the event loop.
        self._async_queue = asyncio.Queue()
        self._asyncio_event_loop = asyncio.new_event_loop()

        self.initialize()

    def initialize(self) -> None:
        """Initialize the GUI widgets"""

        self.setWindowTitle("PySide with asyncio server and client")

        # Create layout
        main_layout = QVBoxLayout()
        self.setLayout(main_layout)

        button_start_server = QPushButton(text="Start server")
        line_edit_message = QLineEdit()
        button_send_message = QPushButton(text="Send message")

        main_layout.addWidget(button_start_server)
        main_layout.addWidget(line_edit_message)
        main_layout.addWidget(button_send_message)

        button_start_server.pressed.connect(
            lambda: self.send_message_to_event_loop("start_server")
        )
        button_send_message.pressed.connect(
            lambda: self.send_message_to_event_loop(line_edit_message.text())
        )

        # Disable the user being able to resize the window by setting a fixed size
        self.setFixedWidth(500)
        self.setFixedHeight(200)

        # Show the window
        self.show()

    def send_message_to_event_loop(self, message: str) -> None:
        """Send the `asyncio` event loop's queue a message by using the coroutine
        `put` and sending it to run on the `asyncio` event loop, putting the message
        on the queue inside the event loop. This must be done because `asyncio.Queue`
        is not threadsafe.
        """
        asyncio.run_coroutine_threadsafe(
            coro=self._async_queue.put(message),
            loop=self._asyncio_event_loop,
        )


async def handle_client(
    reader: asyncio.StreamReader, writer: asyncio.StreamWriter
) -> None:
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info("peername")

    print(f"Received {message!r} from {addr!r}")

    writer.close()
    await writer.wait_closed()


async def run_server():
    server = await asyncio.start_server(handle_client, "127.0.0.1", 8888)
    async with server:
        print("Started server")
        await server.serve_forever()


async def send_message_to_server(message: str) -> None:
    # Lazily create a new connection every time just for demonstration
    # purposes
    _, writer = await asyncio.open_connection("127.0.0.1", 8888)

    writer.write(message.encode())
    await writer.drain()

    writer.close()
    await writer.wait_closed()


async def read_messages(queue: asyncio.Queue) -> None:
    server_task = None
    while True:
        message = await queue.get()
        match message:
            case "start_server":
                server_task = asyncio.create_task(run_server())
            case msg:
                await send_message_to_server(msg)


async def async_main(queue: asyncio.Queue):
    # Launch the tasks to sit around and listen to messages
    await asyncio.gather(read_messages(queue))


def start_asyncio_event_loop(loop: asyncio.AbstractEventLoop) -> None:
    """Starts the given `asyncio` loop on whatever the current thread is"""
    asyncio.set_event_loop(loop)
    loop.set_debug(enabled=True)
    loop.run_forever()


def run_event_loop(queue: asyncio.Queue, loop: asyncio.AbstractEventLoop) -> None:
    """Runs the given `asyncio` loop on a separate thread, passing the queue to the
    event loop for any other thread to send messages to the event loop. The main
    coroutine that is launched on the event loop is `async_main`.
    """
    thread = Thread(target=start_asyncio_event_loop, args=(loop,), daemon=True)
    thread.start()

    asyncio.run_coroutine_threadsafe(async_main(queue), loop=loop)


def run_application(application: QApplication):
    application.exec()


if __name__ == "__main__":
    application = QApplication(sys.argv)
    window = MainWindow()
    async_queue = window._async_queue
    asyncio_event_loop = window._asyncio_event_loop

    run_event_loop(queue=async_queue, loop=asyncio_event_loop)
    sys.exit(run_application(application))