Asyncio＆限速

Question

我正在編寫一個基於asyncio框架的應用程序。這個應用程序與一個具有速率限制的API交互（每秒最多2次）。所以我將與API交互的方法移到了芹菜上，以此作爲速率限制器。但它看起來像一個開銷。

有什麼方法可以創建一個新的asyncio事件循環（或其他），以保證每秒不超過n的coroutins的執行？

Answer 1

我相信你可以寫一個循環是這樣的：

while True: 
    t0 = loop.time() 
    await make_io_call() 
    dt = loop.time() - t0 
    if dt < 0.5: 
     await asyncio.sleep(0.5 - dt, loop=loop) 

Answer 2

接受的答案是正確的。不過請注意，通常情況下，人們希望儘可能接近2QPS。此方法不提供任何並行化，如果make_io_call（）花費超過一秒的時間執行，這可能會成爲問題。一個更好的解決方案是將一個信號傳遞給make_io_call，它可以用來知道它是否可以開始執行。

這是一個這樣的實現：RateLimitingSemaphore只會在速率限制低於要求時纔會釋放它的上下文。

import asyncio 
from collections import deque 
from datetime import datetime 

class RateLimitingSemaphore: 
    def __init__(self, qps_limit, loop=None): 
     self.loop = loop or asyncio.get_event_loop() 
     self.qps_limit = qps_limit 

     # The number of calls that are queued up, waiting for their turn. 
     self.queued_calls = 0 

     # The times of the last N executions, where N=qps_limit - this should allow us to calculate the QPS within the 
     # last ~ second. Note that this also allows us to schedule the first N executions immediately. 
     self.call_times = deque() 

    async def __aenter__(self): 
     self.queued_calls += 1 
     while True: 
      cur_rate = 0 
      if len(self.call_times) == self.qps_limit: 
       cur_rate = len(self.call_times)/(self.loop.time() - self.call_times[0]) 
      if cur_rate < self.qps_limit: 
       break 
      interval = 1./self.qps_limit 
      elapsed_time = self.loop.time() - self.call_times[-1] 
      await asyncio.sleep(self.queued_calls * interval - elapsed_time) 
     self.queued_calls -= 1 

     if len(self.call_times) == self.qps_limit: 
      self.call_times.popleft() 
     self.call_times.append(self.loop.time()) 

    async def __aexit__(self, exc_type, exc, tb): 
     pass 


async def test(qps): 
    executions = 0 
    async def io_operation(semaphore): 
     async with semaphore: 
      nonlocal executions 
      executions += 1 

    semaphore = RateLimitingSemaphore(qps) 
    start = datetime.now() 
    await asyncio.wait([io_operation(semaphore) for i in range(5*qps)]) 
    dt = (datetime.now() - start).total_seconds() 
    print('Desired QPS:', qps, 'Achieved QPS:', executions/dt) 

if __name__ == "__main__": 
    asyncio.get_event_loop().run_until_complete(test(100)) 
    asyncio.get_event_loop().close() 

將打印Desired QPS: 100 Achieved QPS: 99.82723898022084