同时多个异步请求 [英] Multiple async requests simultaneously

问题描述

我正在尝试同时调用 ~ 300 个 API 调用，以便最多在几秒钟内获得结果.

I'm trying to call ~ 300 API calls at the same time, so that I would get the results in a couple of seconds max.

我的伪代码如下:

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 并且代码需要永远，所以我很确定它们不会同时运行.我做错了什么?

Doing this I get getting color printed out every second or so and the code takes forever, so I'm pretty sure they don't run simultaneously. What am I doing wrong?

推荐答案

aiohttp with Native Coroutines (async/await)

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

aiohttp with Native Coroutines (async/await)

Here is a typical pattern that accomplishes what you're trying to do. (Python 3.7+.)

一个主要变化是您需要从为同步 IO 构建的 requests 移动到诸如 aiohttp 专为与 async/await(原生协程)一起使用而构建:

One major change is that you will need to move from requests, which is built for synchronous IO, to a package such as aiohttp that is built specifically to work with async/await (native coroutines):

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+

这有两个不同的元素，一个是协程的异步方面，另一个是在指定任务容器(期货)时引入的并发性:

There are two distinct elements to this, one being the asynchronous aspect of the coroutines and one being the concurrency introduced on top of that when you specify a container of tasks (futures):

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

• You create one coroutine get that uses await with two awaitables: the first being .request and the second being .json. This is the async aspect. The purpose of awaiting these IO-bound responses is to tell the event loop that other get() calls can take turns running through that same routine.
• The concurrent aspect is encapsulated in await asyncio.gather(*tasks). This maps the awaitable get() call to each of your colors. The result is an aggregate list of returned values. Note that this wrapper will wait until all of your responses come in and call .json(). If, alternatively, you want to process them greedily as they are ready, you can loop over asyncio.as_completed: each Future object returned represents the earliest result from the set of the remaining awaitables.

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

Lastly, take note that asyncio.run() is a high-level "porcelain" function introduced in Python 3.7. In earlier versions, you can mimic it (roughly) like:

# 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()

<小时>

限制请求

有多种方法可以限制并发率.例如，请参阅 async-await 函数中的 asyncio.semaphore 或 并发有限的大量任务.

这篇关于同时多个异步请求的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！