python - 等待方法中的异步超时解释

Question

>>> import asyncio
>>> help(asyncio.wait)

..

Help on function wait in module asyncio.tasks:

wait(fs, *, loop=None, timeout=None, return_when='ALL_COMPLETED')
    Wait for the Futures and coroutines given by fs to complete.

    Coroutines will be wrapped in Tasks.

    Returns two sets of Future: (done, pending).

    Usage:

        done, pending = yield from asyncio.wait(fs)

    Note: This does not raise TimeoutError! Futures that aren't done
    when the timeout occurs are returned in the second set.
(END)

我不太明白这个帮助中的最后一个注释(什么是第二组？它是待定/重新处理集吗？我如何执行待定任务并将完成和待定的结果合并然后保存到数据库中)

我的问题: 我将 asyncio 与 aiohttp 一起使用，有数百万个 url，其中很少有人可能会引发超时错误。我想将它们发送到队列中进行重新处理，或者应该由事件池处理。

import asyncio
import aiohttp
sem = asyncio.Semaphore(10)

def process_data(url):
    with (yield from sem):
          response = yield from aiohttp.request('GET', url)
          print(response)

loop = asyncio.get_event_loop()
c = asyncio.wait([process_data(url) for url in url_list], timeout=10)
loop.run_until_complete(c)

PS:我没有使用wait_for方法。

Answer 1

这是帮助中的两组:

Returns two sets of Future: (done, pending).

第二组是pending 组，在超时时间内还没有完成的作业。它将返回一个包含两个 future 列表的元组，一个是已完成的列表，另一个是尚未完成的列表。

代替:

c = asyncio.wait([process_data(url) for url in url_list], timeout=10)
loop.run_until_complete(c)

你可能应该:

def dostuff():
    done, pending = yield from asyncio.wait([process_data(url) for url in url_list], timeout=10)

    # do something with pending

loop.run_until_complete(dostuff())

这里是更多信息:

https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.wait