如何正确使用 asyncio.FIRST_COMPLETED

Question

问题是，RuntimeError: Event loop is closed即使我使用return_when=asyncio.FIRST_COMPLETEDinside ，我也会不断出错await asyncio.wait()。

我的代码：

async def task_manager():
    tasks = [grab_proxy() for _ in range(10)]
    finished, unfinished = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    for x in finished:
        result = x.result()

        if result:
            return result


def get_proxy_loop():
    loop = asyncio.new_event_loop()

    proxy = loop.run_until_complete(task_manager())

    loop.close()
    return proxy


if __name__ == '__main__':
    p = get_proxy_loop()

    print(type(p))
    print(p)

预期行为：

return_when=asyncio.FIRST_COMPLETED 当第一个结果返回“幕后”时，应该终止所有剩余的任务。

但实际上在返回第一个结果后仍然存在未完成的任务。在我关闭循环get_proxy_loop()并访问__main__这些剩余任务中的结果raise 之后RuntimeError: Event loop is closed。

控制台输出：

<class 'str'>
78.32.35.21:55075
Task was destroyed but it is pending!
task: <Task pending coro=<grab_proxy() running at /home/pata/PycharmProjects/accs_farm/accs_farm/proxy_grabber.py:187> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fc5187a8798>()]>>
Exception ignored in: <coroutine object grab_proxy at 0x7fc5150aae60>
Traceback (most recent call last):
  File "/home/pata/proxy_grabber.py", line 187, in grab_proxy
    proxy = await async_get_proxy()
  File "/home/pata/proxy_grabber.py", line 138, in async_get_proxy
    async with session.get(provider_url, timeout=5, params=params) as r:
  File "/home/pata/venvs/test_celery/lib/python3.6/site-packages/aiohttp/client.py", line 855, in __aenter__
    self._resp = await self._coro
  File "/home/pata/venvs/test_celery/lib/python3.6/site-packages/aiohttp/client.py", line 396, in _request
    conn.close()
  File "/home/pata/venvs/test_celery/lib/python3.6/site-packages/aiohttp/connector.py", line 110, in close
    self._key, self._protocol, should_close=True)
  File "/home/pata/venvs/test_celery/lib/python3.6/site-packages/aiohttp/connector.py", line 547, in _release
Event loop is closed
    transport = protocol.close()
  File "/home/pata/venvs/test_celery/lib/python3.6/site-packages/aiohttp/client_proto.py", line 54, in close
    transport.close()
  File "/usr/lib/python3.6/asyncio/selector_events.py", line 621, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "/usr/lib/python3.6/asyncio/base_events.py", line 580, in call_soon
    self._check_closed()
  File "/usr/lib/python3.6/asyncio/base_events.py", line 366, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
...
...
Task was destroyed but it is pending!
task: <Task pending coro=<grab_proxy() done, defined at /home/pata/proxy_grabber.py:183> wait_for=<Future pending cb=[BaseSelectorEventLoop._sock_connect_done(11)(), <TaskWakeupMethWrapper object at 0x7fc514d15e28>()]>>
Task was destroyed but it is pending!
task: <Task pending coro=<grab_proxy() done, defined at /proxy_grabber.py:183> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fc5187a8588>()]>>
Event loop is closed
Process finished with exit code 0

Answer 1

该asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)协程返回时至少一个的任务已经完成。其他任务仍可处于活动状态。这是不是的工作asyncio.wait()取消这些任务给你。的用例asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)是让您监视任务并在完成时根据结果​​采取行动；您通常会反复调用它，直到完成所有任务。

从asyncio.wait()文档：

同时运行aws集中的可等待对象并阻塞，直到return_when指定的条件。

[...]

return_when指示此函数应何时返回。它必须是以下常量之一：

FIRST_COMPLETED
当任何未来完成或取消时，该函数将返回。

[...]

与 不同wait_for()，wait()发生超时时不会取消期货。

文档明确指出它不会取消期货，即使您设置了超时（如果您确实设置了超时，那么第一个完成的集合就是空的，任务仍然处于活动状态并列在第二个待处理集合中）。

如果您需要取消未完成的任务，请明确这样做：

while tasks:
    finished, unfinished = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    for x in finished:
        result = x.result()

        if result:
            # cancel the other tasks, we have a result. We need to wait for the cancellations
            # to propagate.
            for task in unfinished:
                task.cancel()
            await asyncio.wait(unfinished)
            return result

    tasks = unfinished

带有一些额外打印和随机任务的演示：

>>> import asyncio
>>> import random
>>> async def grab_proxy(taskid):
...     await asyncio.sleep(random.uniform(0.1, 1))
...     result = random.choice([None, None, None, 'result'])
...     print(f'Task #{taskid} producing result {result!r}')
...     return result
...
>>> async def task_manager():
...     tasks = [grab_proxy(i) for i in range(10)]
...     while tasks:
...         finished, unfinished = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
...         for x in finished:
...             result = x.result()
...             print(f"Finished task produced {result!r}")
...             if result:
...                 # cancel the other tasks, we have a result. We need to wait for the cancellations
...                 # to propagate.
...                 print(f"Cancelling {len(unfinished)} remaining tasks")
...                 for task in unfinished:
...                     task.cancel()
...                 await asyncio.wait(unfinished)
...                 return result
...         tasks = unfinished
...
>>>
>>> def get_proxy_loop():
...     loop = asyncio.new_event_loop()
...     proxy = loop.run_until_complete(task_manager())
...     loop.close()
...     return proxy
...
>>> get_proxy_loop()
Task #7 producing result None
Finished task produced None
Task #0 producing result 'result'
Finished task produced 'result'
Cancelling 8 remaining tasks
'result'