python - 同时进行多个异步请求

Question

我试图同时调用 ~ 300 个 API 调用，这样我最多可以在几秒钟内得到结果。

我的伪代码如下所示:

def function_1():
    colors = ['yellow', 'green', 'blue', + ~300 other ones]
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    res = loop.run_until_complete(get_color_info(colors))

async def get_color_info(colors):
    loop = asyncio.get_event_loop()
    responses = []
    for color in colors:
        print("getting color")
        url = "https://api.com/{}/".format(color)
        data = loop.run_in_executor(None, requests.get, url)
        r = await data
        responses.append(r.json())
    return responses

这样做我得到 getting color 每隔一秒左右打印一次并且代码需要永远，所以我很确定它们不会同时运行。我做错了什么？

Answer 1

aiohttp 与原生协程(async/await)

这是一个典型的模式，可以完成您想要做的事情。 (Python 3.7+。)

一个主要的变化是您需要从为同步 IO 构建的 requests 移动到诸如 aiohttp 之类的包。这是专门为与 async/await( native 协程)一起工作而构建的:

import asyncio
import aiohttp  # pip install aiohttp aiodns


async def get(
    session: aiohttp.ClientSession,
    color: str,
    **kwargs
) -> dict:
    url = f"https://api.com/{color}/"
    print(f"Requesting {url}")
    resp = await session.request('GET', url=url, **kwargs)
    # Note that this may raise an exception for non-2xx responses
    # You can either handle that here, or pass the exception through
    data = await resp.json()
    print(f"Received data for {url}")
    return data


async def main(colors, **kwargs):
    # Asynchronous context manager.  Prefer this rather
    # than using a different session for each GET request
    async with aiohttp.ClientSession() as session:
        tasks = []
        for c in colors:
            tasks.append(get(session=session, color=c, **kwargs))
        # asyncio.gather() will wait on the entire task set to be
        # completed.  If you want to process results greedily as they come in,
        # loop over asyncio.as_completed()
        htmls = await asyncio.gather(*tasks, return_exceptions=True)
        return htmls


if __name__ == '__main__':
    colors = ['red', 'blue', 'green']  # ...
    # Either take colors from stdin or make some default here
    asyncio.run(main(colors))  # Python 3.7+

这有两个不同的元素，一个是协程的异步方面，一个是当您指定任务容器( future )时在此之上引入的并发性:

• 您创建一个协程 get，它使用 await 和两个 awaitables:第一个是 .request第二个是.json .这是异步方面。 await这些 IO 绑定(bind)响应的目的是告诉事件循环其他 get() 调用可以轮流运行相同的例程。
• 并发方面封装在await asyncio.gather(*tasks)中。这会将可等待的 get() 调用映射到您的每个 colors。结果是返回值的聚合列表。请注意，此包装器将等到您的所有 响应都进入并调用.json()。或者，如果你想在它们准备好时贪婪地处理它们，你可以循环 asyncio.as_completed :返回的每个 Future 对象代表剩余可等待对象集合中最早的结果。

最后，请注意 asyncio.run()是 Python 3.7 中引入的高级“瓷器”函数。在早期版本中，您可以(大致)模仿它:

# The "full" versions makes a new event loop and calls
# loop.shutdown_asyncgens(), see link above
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main(colors))
finally:
    loop.close()

---

限制请求

有多种方法可以限制并发率。例如，参见 asyncio.semaphore in async-await function或 large numbers of tasks with limited concurrency .