无法从正在运行的事件循环中调用 asyncio.run()

Posted 2023-03-30

【中文标题】无法从正在运行的事件循环中调用 asyncio.run()

【英文标题】：asyncio.run() cannot be called from a running event loop

【发布时间】：2019-03-29 02:24:30

【问题描述】：

我想使用 asyncio 来获取网页 html。

我在 jupyter notebook 中运行以下代码：

import aiofiles
import aiohttp
from aiohttp import ClientSession

async def get_info(url, session):
    resp = await session.request(method="GET", url=url)
    resp.raise_for_status()
    html = await resp.text(encoding='GB18030')
    with open('test_asyncio.html', 'w', encoding='utf-8-sig') as f:
        f.write(html)
    return html
    
async def main(urls):
    async with ClientSession() as session:
        tasks = [get_info(url, session) for url in urls]
        return await asyncio.gather(*tasks)

if __name__ == "__main__":
    url = ['http://huanyuntianxiazh.fang.com/house/1010123799/housedetail.htm', 'http://zhaoshangyonghefu010.fang.com/house/1010126863/housedetail.htm']
    result = asyncio.run(main(url))

但是，它返回RuntimeError: asyncio.run() cannot be called from a running event loop

有什么问题？

如何解决？

【参考方案1】：

asyncio.run() 文档说：

这个函数cannot在同一个线程中运行另一个异步事件循环时被调用。

在您的情况下，jupyter (IPython ≥ 7.0) 已经在运行事件循环：

您现在可以在 IPython 终端和笔记本的顶层使用 async/await，它应该 — 在大多数情况下 — “正常工作”。将 IPython 更新到 7+ 版，将 IPykernel 更新到 5+ 版，然后你就可以参加比赛了。

因此，您无需自己启动事件循环，而是可以直接调用await main(url)，即使您的代码位于任何异步函数之外。

Jupyter / IPython

async def main():
    print(1)
    
await main()

Python (≥ 3.7)

import asyncio

async def main():
    print(1)
    
asyncio.run(main())

在您的代码中：

url = ['url1', 'url2']
result = await main(url)

for text in result:
    pass # text contains your html (text) response

注意

有一个slight difference 说明 Jupyter 与 IPython 相比如何使用循环。

【讨论】：

谢谢你，cglacet。但是，有一个警告：c:\program files\python37\lib\site-packages\ipykernel_launcher.py:29: RuntimeWarning: coroutine 'main' was never awaited

这可能是因为您调用了main(url) 而不是await main(url)。

有没有办法让代码 sn-p 在 Jupyter 内部和外部都可以工作？

我不知道这是否可能，但我不确定您是否真的想要那样。也许将此作为​​一个单独的问题提出，看看人们是否对它是否可能或可取有想法。

有趣的是，当我在 jupyter 中运行上述代码时，我得到： SyntaxError: 'await' outside function

【参考方案2】：

要添加到cglacet 的答案 - 如果要检测循环是否正在运行并自动调整（即在现有循环上运行main()，否则asyncio.run()），这里是一个sn-p可能有用：

# async def main():
#     ...

try:
    loop = asyncio.get_running_loop()
except RuntimeError:  # 'RuntimeError: There is no current event loop...'
    loop = None

if loop and loop.is_running():
    print('Async event loop already running. Adding coroutine to the event loop.')
    tsk = loop.create_task(main())
    # ^-- https://docs.python.org/3/library/asyncio-task.html#task-object
    # Optionally, a callback function can be executed when the coroutine completes
    tsk.add_done_callback(
        lambda t: print(f'Task done with result=t.result()  << return val of main()'))
else:
    print('Starting new event loop')
    asyncio.run(main())

【讨论】：

很好，确实在这里添加回调可能是获取结果的唯一方法。在函数中包含此代码可能会很有趣，因此您只需调用类似run(main(*args, **kwargs)) 的东西。该函数还可以确保您恢复输出，但不确定这是否真的很简单（因为回调）。

经过一些实验后，处理结果的唯一方法似乎是在异步代码中（如果需要进一步嵌套和结构化异步函数和回调）。任何在异步函数之外等待结果的尝试（= 在 同步 代码中）都会阻塞整个事情（如果它在同一进程中运行）。

【参考方案3】：

就用这个吧：

https://github.com/erdewit/nest_asyncio

import nest_asyncio
nest_asyncio.apply()

【讨论】：

nest_asyncio 的文档提到了 issue report with asyncio，其中明确指出这不是 asyncio 的预期行为。所以我会认为nest_asyncio 是一个我不相信在我的代码库中不会用新的 Python 版本破坏一切的大胖黑客。

另一个问题是nest_asyncio需要Python 3.5或更高版本，这对我没有帮助，因为我被困在Python 2.7上。

这让我可以解决导入代码的问题（您不能自己更改）

【参考方案4】：

结合 Pankaj Sharma 和 Jean Monet 的方法，我编写了以下 sn-p，它充当 asyncio.run（语法略有不同），但也可以在 Jupyter notebook 中工作。

class RunThread(threading.Thread):
    def __init__(self, func, args, kwargs):
        self.func = func
        self.args = args
        self.kwargs = kwargs
        super().__init__()

    def run(self):
        self.result = asyncio.run(self.func(*self.args, **self.kwargs))

def run_async(func, *args, **kwargs):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None
    if loop and loop.is_running():
        thread = RunThread(func, args, kwargs)
        thread.start()
        thread.join()
        return thread.result
    else:
        return asyncio.run(func(*args, **kwargs))

用法：

async def test(name):
    await asyncio.sleep(5)
    return f"hello name"

run_async(test, "user")  # blocks for 5 seconds and returns "hello user"

【讨论】：

感谢您提供这个聪明的解决方案。您对“asyncio 附加到不同的循环”错误有什么建议吗？在子循环内部，我需要使用主循环中的 2 个协程。

【参考方案5】：

我发现unsync 包对于编写在 Python 脚本和 Jupyter REPL 中行为相同的代码很有用。

import asyncio
from unsync import unsync


@unsync
async def demo_async_fn():
    await asyncio.sleep(0.1)
    return "done!"

print(demo_async_fn().result())

【参考方案6】：

正如 cglacet 提到的，文档中说

当另一个异步事件循环发生时，无法调用此函数 在同一个线程中运行。

您可以使用另一个线程，即 -

class ResolveThread(threading.Thread):
            def __init__(self,result1,fun,url):
                self.result1= result1
                self.fun = fun
                self.url = url
                threading.Thread.__init__(self)
            def run(self):
                result1[0] = asyncio.run(self.fun(self.url))


result1 = [None]
sp = ResolveThread(result1)
sp.start()
sp.join() # connect main thread
result = result1[0]