Python asyncio - 增加 Semaphore 的值答案

【问题标题】：Python asyncio - Increase the value of SemaphorePython asyncio - 增加 Semaphore 的值
【发布时间】：2021-12-04 19:10:16
【问题描述】：

我在我的一个项目中使用aiohttp，并希望限制每秒发出的请求数。我正在使用asyncio.Semaphore 来做到这一点。我的挑战是我可能想增加/减少每秒允许的请求数。

例如：

limit = asyncio.Semaphore(10)
async with limit:
    async with aiohttp.request(...)
        ...
    await asyncio.sleep(1)

这很好用。也就是说，它将aiohttp.request 限制为每秒 10 个并发请求。但是，我可能想增加和减少Semaphore._value。我可以做到limit._value = 20，但我不确定这是正确的方法还是有其他方法。

【问题讨论】：

标签： python-asyncio

【解决方案1】：

访问私有 _value 属性不是正确的方法，至少有两个原因：一个是该属性是私有的，可以在未来版本中删除、重命名或更改含义，恕不另行通知，另一个是增加已经有服务员的信号量不会注意到 limit。

由于asyncio.Semaphore 不支持动态修改限制，您有两种选择：实现自己的支持它的Semaphore 类，或者根本不使用Semaphore。后者可能更容易，因为您总是可以用固定数量的工作任务替换信号量强制限制，这些工作任务通过队列接收作业。假设您当前的代码如下所示：

async def fetch(limit, arg):
    async with limit:
        # your actual code here
        return result

async def tweak_limit(limit):
    # here you'd like to be able to increase the limit

async def main():
    limit = asyncio.Semaphore(10)
    asyncio.create_task(tweak_limit(limit))
    results = await asyncio.gather(*[fetch(limit, x) for x in range(1000)])

您可以通过提前创建工作人员并让他们有工作去做来表达它而无需信号量：

async def fetch_task(queue, results):
    while True:
        arg = await queue.get()
        # your actual code here
        results.append(result)
        queue.task_done()

async def main():
    # fill the queue with jobs for the workers
    queue = asyncio.Queue()
    for x in range(1000):
        await queue.put(x)

    # create the initial pool of workers
    results = []
    workers = [asyncio.create_task(fetch_task(queue, results))
               for _ in range(10)]
    asyncio.create_task(tweak_limit(workers, queue, results))

    # wait for workers to process the entire queue
    await queue.join()
    # finally, cancel the now-idle worker tasks
    for w in workers:
        w.cancel()

    # results are now available

tweak_limit() 函数现在可以通过生成新工作人员来增加限制：

async def tweak_limit(workers, queue, results):
    while True:
        await asyncio.sleep(1)
        if need_more_workers:
            workers.append(asyncio.create_task(fetch_task(queue, results)))

【讨论】：

【解决方案2】：

使用工作者和队列是一个更复杂的解决方案，你必须考虑诸如设置、拆卸、异常处理和背压等问题。

信号量可以用 Lock 来实现，如果你不介意效率低下（你会明白为什么），这里有一个动态值信号量的简单实现：

class DynamicSemaphore:
    def __init__(self, value=1):
        self._lock = asyncio.Lock()

        if value < 0:
            raise ValueError("Semaphore initial value must be >= 0")

        self.value = value

    async def __aenter__(self):
        await self.acquire()
        return None

    async def __aexit__(self, exc_type, exc, tb):
        self.release()

    def locked(self):
        return self.value == 0

    async def acquire(self):
        async with self._lock:
            while self.value <= 0:
                await asyncio.sleep(0.1)

        self.value -= 1
        return True

    def release(self):
        self.value += 1

【讨论】：