为什么 asyncio.sleep() 会导致“将 Future 附加到不同的循环”？

Question

我一直在尝试理解 Python 中的异步编程。我尝试编写一个简单的节流器，它一次只允许处理 rate_limit 个任务。

下面是实现：

import asyncio
import time

class Throttler:

    def __init__ (
            self,
            rate_limit: int,
            retry_interval: float
    ) -> None:

        self.rate_limit = rate_limit
        self.retry_interval = retry_interval

        self._time_counter = None
        self._tasks_counter = 0

        self.lock = asyncio.Lock()

    async def __aenter__(
            self
    ) -> 'Throttler':

        async with self.lock:
            print(f'Starting {self._tasks_counter}')

            if self._time_counter is None:
                pass
            else:
                difference = time.perf_counter() - self._time_counter
                # if difference < self.retry_interval:
                #     await asyncio.sleep(self.retry_interval - difference)

            while True:
                if self._tasks_counter < self.rate_limit:
                    break
                else:
                    print('here')
                    await asyncio.sleep(self.retry_interval)

            if self._time_counter is not None:
                print(time.perf_counter() - self._time_counter)

            self._time_counter = time.perf_counter()
            self._tasks_counter += 1

        return self

    async def __aexit__(
            self,
            exc_type,
            exc_val,
            exc_tb
    ) -> None:

        async with self.lock:
            self._tasks_counter -= 1
            print(f'Ending {self._tasks_counter}')

throttler = Throttler(rate_limit = 5, retry_interval = 2.0)

async def f ():
    async with throttler:
        print(42)
        await asyncio.sleep(1)

async def main ():
    await asyncio.gather(*[f() for i in range(10)])

asyncio.run(main())

我预计当我声明 Throttler 时 rate_limit 设置为 5，它一次最多应处理 5 个请求，然后等待其中一个或多个完成开始处理其他请求。但它并没有像我预期的那样工作，并在遇到 asyncio.sleep 语句之一时引发 RuntimeError（如果您取消注释，即使是已注释的语句）。

这是完整的回溯：

Traceback (most recent call last):
  File "C:\Users\Aryan V S\Desktop\Projects\General\Python\Other\Async\throttle.py", line 288, in f
    await asyncio.sleep(1)
  File "C:\Programming\Python\lib\asyncio\locks.py", line 120, in acquire
    await fut
RuntimeError: Task <Task pending name='Task-3' coro=<f() running at C:\Users\Aryan V S\Desktop\Projects\General\Python\Other\Async\throttle.py:288> cb=[gather.<locals>._done_callback() at C:\Programming\Python\lib\asyncio\tasks.py:766, gather.<locals>._done_callback() at C:\Programming\Python\lib\asyncio\tasks.py:766]> got Future <Future pending> attached to a different loop
Traceback (most recent call last):
  File "C:\Users\Aryan V S\Desktop\Projects\General\Python\Other\Async\throttle.py", line 293, in <module>
    asyncio.run(main())
  File "C:\Programming\Python\lib\asyncio\runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "C:\Programming\Python\lib\asyncio\base_events.py", line 642, in run_until_complete
    return future.result()
  File "C:\Users\Aryan V S\Desktop\Projects\General\Python\Other\Async\throttle.py", line 291, in main
    await asyncio.gather(*[f() for i in range(10)])
  File "C:\Users\Aryan V S\Desktop\Projects\General\Python\Other\Async\throttle.py", line 286, in f
    async with throttler:
  File "C:\Users\Aryan V S\Desktop\Projects\General\Python\Other\Async\throttle.py", line 247, in __aenter__
    async with self.lock:
  File "C:\Programming\Python\lib\asyncio\locks.py", line 14, in __aenter__
    await self.acquire()
  File "C:\Programming\Python\lib\asyncio\locks.py", line 120, in acquire
    await fut
RuntimeError: Task <Task pending name='Task-8' coro=<f() running at C:\Users\Aryan V S\Desktop\Projects\General\Python\Other\Async\throttle.py:286> cb=[gather.<locals>._done_callback() at C:\Programming\Python\lib\asyncio\tasks.py:766]> got Future <Future pending> attached to a different loop

我在这里做错了什么？ asyncio.sleep 不使用当前正在运行的事件循环还是我不明白它是如何工作的？非常感谢您的宝贵时间！

Answer 1

这是因为您在 asyncio.run 运行循环之前创建了锁，请尝试在 main() 中创建节流器并将其发送到 f()。