python3 异步错误 asyncio.Semaphore RuntimeError: Task got Future attached to a different loop

水深无声 2021-08-31 01:04 667阅读 0赞

### 错误现象 ###

asyncio.Semaphore RuntimeError: Task got Future attached to a different loop

asyncio.Semaphore RuntimeError：任务将Future连接到另一个循环

当我在Python 3.7中运行此代码时：

import asyncio
    
    sem = asyncio.Semaphore(2)
    
    async def work():
        async with sem:
            print('working')
            await asyncio.sleep(1)
    
    async def main():
        await asyncio.gather(work(), work(), work())
    
    asyncio.run(main())

它因RuntimeError失败：

$ python3 demo.py
    working
    working
    Traceback (most recent call last):
      File "demo.py", line 13, in <module>
        asyncio.run(main())
      File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/runners.py", line 43, in run
        return loop.run_until_complete(main)
      File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
        return future.result()
      File "demo.py", line 11, in main
        await asyncio.gather(work(), work(), work())
      File "demo.py", line 6, in work
        async with sem:
      File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/locks.py", line 92, in __aenter__
        await self.acquire()
      File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/locks.py", line 474, in acquire
        await fut
    RuntimeError: Task <Task pending coro=<work() running at demo.py:6> cb=[gather.<locals>._done_callback() at /opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/tasks.py:664]> got Future <Future pending> attached to a different loop

### 错误原因 ###

这是因为Semaphore构造函数在[asyncio / locks.py中][asyncio _ locks.py]设置了`_loop`属性：

class Semaphore(_ContextManagerMixin):
    
        def __init__(self, value=1, *, loop=None):
            if value < 0:
                raise ValueError("Semaphore initial value must be >= 0")
            self._value = value
            self._waiters = collections.deque()
            if loop is not None:
                self._loop = loop
            else:
                self._loop = events.get_event_loop()

但是`asyncio.run()`启动了一个全新的循环–在[asyncio / runners.py中][asyncio _ runners.py] ，它在文档中也有[提及][asyncio _ runners.py] ：

def run(main, *, debug=False):
        if events._get_running_loop() is not None:
            raise RuntimeError(
                "asyncio.run() cannot be called from a running event loop")
    
        if not coroutines.iscoroutine(main):
            raise ValueError("a coroutine was expected, got {!r}".format(main))
    
        loop = events.new_event_loop()
        ...

在`asyncio.run()`外部启动的`Semaphore`将获取asyncio“默认”循环，因此不能与通过`asyncio.run()`创建的事件循环一起使用。

### 解决方法 ###

从`asyncio.run()`调用的代码中启动`Semaphore` 。 您将必须将它们传递到正确的位置，还有更多的方法可以做到这一点，例如，可以使用[contextvars][] ，但是我仅给出最简单的示例：

import asyncio
    
    async def work(sem):
        async with sem:
            print('working')
            await asyncio.sleep(1)
    
    async def main():
        sem = asyncio.Semaphore(2)
        await asyncio.gather(work(sem), work(sem), work(sem))
    
    asyncio.run(main())

有多个 asyncio loop 时，在创建 asyncio.Semaphore 时需要指定使用哪个 loop，例如 asyncio.Semaphore(2, loop=xxx\_loop)

[asyncio _ locks.py]: https://stackoom.com/link/aHR0cHM6Ly9naXRodWIuY29tL3B5dGhvbi9jcHl0aG9uL2Jsb2IvM2JiY2M5MjU3N2Y4ZTYxNmJjOTRjNjc5MDQwMDQzYmFjZDAwZWJmMS9MaWIvYXN5bmNpby9sb2Nrcy5weQ==
[asyncio _ runners.py]: https://stackoom.com/link/aHR0cHM6Ly9naXRodWIuY29tL3B5dGhvbi9jcHl0aG9uL2Jsb2IvM2JiY2M5MjU3N2Y4ZTYxNmJjOTRjNjc5MDQwMDQzYmFjZDAwZWJmMS9MaWIvYXN5bmNpby9ydW5uZXJzLnB5
[contextvars]: https://stackoom.com/link/aHR0cHM6Ly9kb2NzLnB5dGhvbi5vcmcvMy9saWJyYXJ5L2NvbnRleHR2YXJzLmh0bWw=