Python Aiohttp Asyncio：如何在每个任务之间创建延迟

Question

我试图解决的问题：我正在向服务器发出许多 api 请求。我试图在异步 api 调用之间创建延迟以符合服务器的速率限制策略。

我想让它做什么我希望它表现得像这样：

1. 发出 api 请求#1
2. 等待0.1秒
3. 发出 api 请求 #2
4. 等待0.1秒 ... 等等 ...
5. 重复直到发出所有请求
6. 收集响应并将结果返回一个对象（结果）

   问题：我什么时候介绍的asyncio.sleep()或者时间.睡眠()在代码中，它仍然几乎是瞬间发出 api 请求。它似乎延迟了执行打印（），但不是 api 请求。我怀疑我必须在环形，而不是在 fetch_one() 或 fetch_all() 处，但不知道该怎么做。

   代码块：

   async def fetch_all(loop, urls, delay): 
       results = await asyncio.gather(*[fetch_one(loop, url, delay) for url in urls], return_exceptions=True)
       return results
   
   async def fetch_one(loop, url, delay):
   
       #time.sleep(delay)
       #asyncio.sleep(delay)
   
       async with aiohttp.ClientSession(loop=loop) as session:
           async with session.get(url, ssl=SSLContext()) as resp:
               # print("An api call to ", url, " is made at ", time.time())
               # print(resp)
               return await resp
   
   delay = 0.1
   urls = ['some string list of urls']
   loop = asyncio.get_event_loop()
   loop.run_until_complete(fetch_all(loop, urls, delay))
   
   Versions I'm using: 
   python                    3.8.5
   aiohttp                   3.7.4
   asyncio                   3.4.3
   

   我将不胜感激指导我走向正确方向的任何提示！

Answer 1

对asyncio.gather 的调用将“同时”启动所有请求——另一方面，如果您只是对每个任务使用锁或等待，那么您根本不会从使用并行性中获得任何好处。

最简单的做法是，如果您知道可以发出请求的速率，只需在每个连续请求之前增加异步暂停时间——一个简单的全局变量可以做到这一点：

next_delay = 0.1

async def fetch_all(loop, urls, delay): 
    results = await asyncio.gather(*[fetch_one(loop, url, delay) for url in urls], return_exceptions=True)
    return results

async def fetch_one(loop, url, delay):
    global next_delay
    
    next_delay += delay
    await asyncio.sleep(next_delay)

    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.get(url, ssl=SSLContext()) as resp:
            # print("An api call to ", url, " is made at ", time.time())
            # print(resp)
            return await resp

delay = 0.1
urls = ['some string list of urls']
loop = asyncio.get_event_loop()
loop.run_until_complete(fetch_all(loop, urls, delay))

现在，如果你想发出 5 个请求，然后发出下 5 个请求，你可以使用像 asyncio.Condition 这样的同步原语，在检查有多少 api 调用处于活动状态的表达式上使用它的 wait_for：

active_calls = 0

MAX_CALLS = 5

async def fetch_all(loop, urls, delay): 
    event = asyncio.Event()
    event.set()
    results = await asyncio.gather(*[fetch_one(loop, url, delay, event) for url in urls], return_exceptions=True)
    return results

async def fetch_one(loop, url, delay, cond):
    global active_calls
    
    active_calls += 1
    if active_calls > MAX_CALLS:
        event.clear()
        
    await event.wait()
    
    try:
        async with aiohttp.ClientSession(loop=loop) as session:
            async with session.get(url, ssl=SSLContext()) as resp:
                # print("An api call to ", url, " is made at ", time.time())
                # print(resp)
                return await resp
    finally:
        active_calls -= 1
    if active_calls == 0:
        event.set()
        

urls = ['some string list of urls']
loop = asyncio.get_event_loop()
loop.run_until_complete(fetch_all(loop, urls, delay))

对于这两个示例，如果您的任务在设计中避免使用全局变量（实际上，这些是“模块”变量） - 您可以将所有功能移动到一个类中，并在一个实例上工作，并将全局变量提升为实例属性，或者使用可变容器，例如在其第一项中保存 active_calls 值的列表，并将其作为参数传递。

Answer 2

当您使用asyncio.gather 时，您同时运行所有fetch_one 协程。他们一起等待delay，而不是一起即时调用 API。

要解决这个问题，您应该在fetch_all 中一个一个地等待fetch_one，或者使用Semaphore 来表示下一个不应该在上一个完成之前开始。

这是想法：

import asyncio

_sem = asyncio.Semaphore(1)


async def fetch_all(loop, urls, delay): 
    results = await asyncio.gather(*[fetch_one(loop, url, delay) for url in urls], return_exceptions=True)
    return results

async def fetch_one(loop, url, delay):

    async with _sem:  # next coroutine(s) will stuck here until the previous is done
        await asyncio.sleep(delay)

        async with aiohttp.ClientSession(loop=loop) as session:
            async with session.get(url, ssl=SSLContext()) as resp:
                # print("An api call to ", url, " is made at ", time.time())
                # print(resp)
                return await resp

delay = 0.1
urls = ['some string list of urls']
loop = asyncio.get_event_loop()
loop.run_until_complete(fetch_all(loop, urls, delay))