在设定的时间间隔后从队列中删除项目答案

【问题标题】：Remove item from Queue after a set interval在设定的时间间隔后从队列中删除项目
【发布时间】：2022-01-22 11:12:33
【问题描述】：

到目前为止，我已经查看了许多（如果不是全部）asyncio / aiohttp 速率限制器。他们都未能解决对我正在尝试使用的 REST API 施加的速率限制。每当我每分钟提供 > 100 个请求时，REST API 都会给我一个 504（服务不可用）。

我尝试限制在一分钟内发送的请求数量，但我仍然得到 504。事实证明，这是因为服务器包含了处理请求所需的时间。然后我尝试根据 received 响应的 nr 限制请求数量。这也不起作用，因为现在，当前 100 个请求返回时，我已经发送了超过 100 个请求。

LeakyBucket 机制不起作用，因为它让我们通过请求 101、102、103.. 在前 100 个已爆破低谷后以稳定的速度。导致每分钟超过 100 个请求。 TokenBucket 机制会减慢一切，即使仅发出 90 个请求时也是如此。

我基本上想做的是；创建一个在发出请求时添加的队列，并在请求完成一分钟后清除队列中的项目。

我喜欢asyncio.Queue 在您尝试put 队列中的新项目时已经实现等待的方式。下面的代码有效，但它使代码保持运行一分钟，即使只发出一个请求。

import asyncio


class Ratelimiter:
    def __init__(self, max_calls: int, period: int):
        self.period = period
        self.queue = asyncio.Queue(maxsize=max_calls)

    async def __aenter__(self) -> "Ratelimiter":
        await self.queue.put(None)
        return self

    async def __aexit__(self, exc_type, exc_value, exc_tb) -> None:
        await asyncio.sleep(self.interval)
        await self.queue.get()
        self.queue.task_done()
        return None

【问题讨论】：

标签： python python-asyncio

【解决方案1】：

我找到了解决方案；通过让一个单独的线程来处理倒数计时器，主线程可以继续。

import asyncio
import collections
import threading
import time


class Ratelimiter:
    def __init__(self, max_calls: int, period: int):
        self.max_calls = max_calls
        self.interval = period
        self.queue = collections.deque()

    async def __aenter__(self) -> "Ratelimiter":
        while len(self.queue) >= self.max_calls:
            await asyncio.sleep(
                self.queue[0] + self.interval - time.monotonic()
            )
        self.queue.append(time.monotonic())
        return self

    async def __aexit__(self, exc_type, exc_value, exc_tb) -> None:
        threading.Thread(target=self._sleep_and_remove).start()
        return None

    def _sleep_and_remove(self):
        time.sleep(self.interval)
        self.queue.popleft()

【讨论】：