使用 Python asyncio 从同步函数中运行并等待异步函数

Question

在我的代码中，我有一个带有属性的类，偶尔需要运行异步代码。有时我需要从异步函数访问属性，有时从同步函数访问属性 - 这就是为什么我不希望我的属性是异步的。此外，我的印象是异步属性通常是一种代码异味。如果我错了纠正我。

我在从同步属性执行异步方法并阻止进一步执行直到异步方法完成时遇到问题。

这是一个示例代码：

import asyncio


async def main():
    print('entering main')
    synchronous_property()
    print('exiting main')


def synchronous_property():
    print('entering synchronous_property')
    loop = asyncio.get_event_loop()
    try:
        # this will raise an exception, so I catch it and ignore
        loop.run_until_complete(asynchronous())
    except RuntimeError:
        pass
    print('exiting synchronous_property')


async def asynchronous():
    print('entering asynchronous')
    print('exiting asynchronous')


asyncio.run(main())

它的输出：

entering main
entering synchronous_property
exiting synchronous_property
exiting main
entering asynchronous
exiting asynchronous

首先，RuntimeError捕获似乎是错误的，但如果我不这样做，我会得到RuntimeError: This event loop is already running例外。

其次，asynchronous()函数最后执行，同步完成后。我想通过异步方法对数据集进行一些处理，所以我需要等待它完成。如果我await asyncio.sleep(0)在调用之后添加synchronous_property()，它会asynchronous()在main()完成之前调用，但这对我没有帮助。我需要asynchronous()在synchronous_property()完成之前运行。

我错过了什么？我正在运行 python 3.7。

Answer 1

Asyncio在设计上确实坚持不允许嵌套事件循环。但是，您始终可以在不同的线程中运行另一个事件循环。这是一个使用线程池的变体，以避免每次都创建一个新线程：

\n

import asyncio, concurrent.futures\n\nasync def main():\n    print(\'entering main\')\n    synchronous_property()\n    print(\'exiting main\')\n\npool = concurrent.futures.ThreadPoolExecutor()\n\ndef synchronous_property():\n    print(\'entering synchronous_property\')\n    result = pool.submit(asyncio.run, asynchronous()).result()\n    print(\'exiting synchronous_property\', result)\n\nasync def asynchronous():\n    print(\'entering asynchronous\')\n    await asyncio.sleep(1)\n    print(\'exiting asynchronous\')\n    return 42\n\nasyncio.run(main())\n

\n

此代码在每个sync\xe2\x86\x92async 边界上创建一个新的事件循环，因此如果您经常这样做，则不要指望高性能。可以通过使用 为每个线程仅创建一个事件循环asyncio.new_event_loop并将其缓存在线程局部变量中来改进它。

\n

Answer 2

最简单的方法是使用现有的“轮子”，例如 asgiref.async_to_sync

from asgiref.sync import async_to_sync

然后：

async_to_sync(main)()

一般来说：

async_to_sync(<your_async_func>)(<.. arguments for async function ..>)

这是一个调用者类，它将仅在具有事件循环的线程上工作的可等待对象转换为在子线程中工作的同步可调用对象。

如果调用堆栈包含异步循环，则代码将在那里运行。否则，代码将在新线程的新循环中运行。

无论哪种方式，该线程都会暂停并等待运行使用 SyncToAsync 从调用堆栈下方调用的任何 thread_sensitive 代码，然后在异步任务返回后最终退出。