并发运行Python异步服务器循环和未来

Question

我正在进行一个项目，其中我有一个简单的套接字服务器，寻找连接来回复到客户端的消息队列中的消息。同时，我有一个使用PyWin32的监视系统，它正在查找目录中文件的更改。当PyWin32检测到文件更改时，它应该将信息附加到消息队列中。当服务器检测到连接时，它应该从消息队列中获取相关信息。服务器将不断查找连接，而PyWin32只在检测到更改时才运行。

我可以分别成功地运行它们，但不能一起运行。我的代码似乎挂在一个函数上，而另一个函数则停止处理。我读过这些文档并仔细阅读了这些帖子，但似乎没有什么能解决我的问题。

下面是套接字服务器的示例：

class TestSocketServer:
    def __init__(self,
                 # response_handler: Callable
                 ):
        self._alive: bool = False

        # Configure client-server connection
        self._host: str = 'localhost'
        self._port: int = 50000

        self._mq = {}

    def _respond(self, reader):
        request = self._mq[reader].get()

        return f"Got request {request}"

    async def _client_connected(self, reader, writer):
        data = await reader.read(1024)

        # print(reader, type(reader))
        message = data.decode()
        print(message)

        self._mq[reader]: queue.Queue = queue.Queue()
        self._mq[reader].put_nowait(message)

        response = self._respond(reader)

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

        writer.close()

    async def handle_traffic(self):
        server = await asyncio.start_server(
            self._client_connected,
            self._host,
            self._port)

        await server.serve_forever()

    def start(self):
        asyncio.run(self.handle_traffic())

PyWin32处理程序如下所示：

    async def _monitor(self):
        """
        Look for changes in directory and return them
        :return: FILE_NOTIFY_INFORMATION
        """
        ReadDirectoryChangesW(self._handle,  # Python version of Windows Handle
                          self._buffer,
                          TRUE,  # Monitor directory tree
                          # Attributes to monitor
                          (FILE_NOTIFY_CHANGE_SIZE |
                           FILE_NOTIFY_CHANGE_ATTRIBUTES |
                           FILE_NOTIFY_CHANGE_DIR_NAME |
                           FILE_NOTIFY_CHANGE_FILE_NAME |
                           FILE_NOTIFY_CHANGE_LAST_WRITE)
                          )

        change = FILE_NOTIFY_INFORMATION(self._buffer, 9999)

        action, item = change[0]
        act_str = EventWatcher._ACTIONS.get(action)

        if change:
            chg_msg = {str(time.time()): [str(act_str), item]}
            # TODO try to push changes through socket before persisting
            await self._write_changes(action=action, change=chg_msg)
    async.run(self._monitor)

我是如何尝试实现它们的(不包括async.run()调用)：

loop = asyncio.new_event_loop()
loop.create_task(self.handle_traffic())
loop.create_task(self.monitor())
loop.run_forever()

我还尝试在循环中为监视部分创建一个未来：

loop = asyncio.new_event_loop()
loop.create_task(self.handle_traffic())
fut = loop.create_future()
loop.create_task(self.monitor(fut))
loop.run_forever()

Answer 1

感谢@thisisalsomypassword指出我的Win32 API调用阻塞了我的脚本的其余部分，并且我需要使用线程。下面的实现解决了我的问题：

class TestSocketServer:
    def __init__(self,
                 # response_handler: Callable
                 ):
        self._alive: bool = False

        # Configure client-server connection
        self._host: str = 'localhost'
        self._port: int = 50000

        self._mq = {}

    def _respond(self, reader):
        request = self._mq[reader].get()

        return f"Got request {request}"

    async def _client_connected(self, reader, writer):
        data = await reader.read(1024)

        # print(reader, type(reader))
       message = data.decode()
        print(message)

        self._mq[reader]: queue.Queue = queue.Queue()
        self._mq[reader].put_nowait(message)

        response = self._respond(reader)

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

        writer.close()

    async def start_serving(self):
        await asyncio.start_server(
            self._client_connected,
            self._host,
            self._port)

对于观察者来说：

    def monitor(self):
        """
        Look for changes in directory and return them
        :return: FILE_NOTIFY_INFORMATION
        """
        while True:
            ReadDirectoryChangesW(self._handle,  # Python version of Windows Handle
                                  self._buffer,
                                  TRUE,  # Monitor directory tree
                                  # Attributes to monitor
                                  (FILE_NOTIFY_CHANGE_SIZE |
                                   FILE_NOTIFY_CHANGE_ATTRIBUTES |
                                FILE_NOTIFY_CHANGE_DIR_NAME |
                                FILE_NOTIFY_CHANGE_FILE_NAME |
                                FILE_NOTIFY_CHANGE_LAST_WRITE)
                                )

            change = FILE_NOTIFY_INFORMATION(self._buffer, 9999)

            action, item = change[0]
            print(change)

然后以这种方式实现这两种方式：

self._server = TestSocketServer()
self._watcher = EventWatcher()

def main(self):
    loop = asyncio.get_event_loop()
    loop.create_task(asyncio.to_thread(self._watcher.monitor))
    loop.create_task(self._server.start_serving())
    loop.run_forever()

main()

现在，每当出现套接字请求或文件系统中的更改时，我都会看到两者都出现，程序将继续处理未来的输入。