为什么asyncio.sleep会触发Future绑定到不同事件循环的错误

你碰到的RuntimeError: Future attached to a different loop和asyncio.sleep本身无关，问题出在异步原语的事件循环绑定逻辑：

• asyncio.Lock这类异步同步原语在实例化时，会自动绑定到当前线程的活跃事件循环上
• 你的Throttler是在全局作用域实例化的，此时asyncio.run()还没执行，程序使用的是初始化阶段的默认空循环
• asyncio.run()启动时会销毁旧循环、创建全新的事件循环，此时之前创建的Lock仍然绑定旧循环，在新循环中使用就会抛出跨循环错误

有两种常用修复方式，更推荐第二种适配性更强：

• 方案1：将Throttler的实例化操作移到main函数内部，此时asyncio.run()已经启动了新循环，Lock会绑定到正确的循环上，再将实例作为参数传入任务函数即可
• 方案2：修改Throttler实现，延迟Lock的初始化时机，等到第一次进入异步上下文时再创建Lock，保证绑定的是当前运行的循环

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
        # 不在初始化阶段创建Lock
        self.lock = None

    async def __aenter__(
            self
    ) -> 'Throttler':
        # 第一次进入异步上下文时再创建Lock，此时已绑定正确的运行循环
        if self.lock is None:
            self.lock = asyncio.Lock()
        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())

修复后程序会完全符合预期：同一时间最多5个任务执行，第一批5个任务执行1秒后退出，第二批5个任务立即启动，总执行时间约2秒。

内容的提问来源于stack exchange，提问作者Arrow