根据条件关闭 Python 中的任务

Question

我正在使用 AsyncIO 和 Websockets 模块在 Python 中创建两个并发任务，每个任务都连接到一个 websocket 服务器并接收消息。

我正在尝试创建一个系统，当一个任务超过 4 秒没有收到消息时，它必须关闭连接和任务，但我很难弄清楚这一点。谁能帮我解决这个问题？

这是我尝试过的：

async def connect(URI):
    async with websockets.client.connect(URI) as ws:
        LastUpdate = time.time()
        while time.time() - LastUpdate < 4:
            LastUpdate = time.time()
            await ws.recv()

        print('Not receiving updates anymore')
                
async def main():
    Tasks = [asyncio.create_task(connect('URI1')), asyncio.create_task(connect('URI2'))]

    await asyncio.gather(*Tasks)

在这里，main 创建了两个并发任务并同时运行它们。 connect 负责连接和接收消息。我试图做的是在最后一次更新超过 4 秒之前打破 while 循环，但这不起作用，因为循环没有被打破。

Answer 1

你可以使用asyncio.wait_for:

async def connect(URI):
    async with websockets.client.connect(URI) as ws:
        while True:
            try:
                msg = await asyncio.wait_for(ws.recv(), 4)
            except asyncio.TimeoutError:
                break
            # do something with msg

        print('Not receiving updates anymore')

Answer 2

有几种方法可以做到这一点。

1. 您可以在异步例程中使用的包中设置超时。
2. 您可以使用 asyncio 并在任务超时时终止它们。
3. 您可以对所有操作使用非阻塞套接字，并且根本不使用 asyncio，而是使用 select() 或 poll()。这非常有效，但要复杂得多。

我会按照这个顺序进行，第一个在大多数情况下是最简单和最喜欢的。 Websockets 似乎可以处理超时：https://github.com/websocket-client/websocket-client/blob/29c15714ac9f5272e1adefc9c99b83420b409f63/websocket/_socket.py#L99

您可能想在套接字上设置超时。通常这是使用https://docs.python.org/3.8/library/socket.html#socket.socket.settimeout

我想 websockets 虽然在某个地方公开了这个功能，但我不熟悉它。