使用 asyncio.run_coroutine_threadsafe 在 Python 中处理异步操作

使用 `asyncio.run_coroutine_threadsafe` 在 Python 中处理异步操作

在现代 Python 开发中，异步编程已经成为一种常见的模式。尤其是在处理 I/O 密集型操作（如网络请求或文件读写）时，异步编程能够显著提高程序的性能。本文将介绍如何使用 asyncio.run_coroutine_threadsafe 方法来在多线程环境中安全地调度异步操作。

什么是 `asyncio.run_coroutine_threadsafe`？

asyncio.run_coroutine_threadsafe 是 asyncio 库中的一个实用函数，它允许在与事件循环不在同一线程的情况下安全地调度异步协程。这对于在多线程程序中使用异步操作非常重要，因为直接从非事件循环线程调用协程会导致错误。

使用场景

想象一个场景：主线程运行同步框架或 ROS 节点，另一个后台线程维护 asyncio 事件循环。同步线程收到请求后，希望把一个协程提交到该事件循环中执行。这里就是 asyncio.run_coroutine_threadsafe 的用武之地。

示例代码

以下是一个示例，展示如何使用 asyncio.run_coroutine_threadsafe 来调度异步操作：

import asyncio
import threading

async def async_task():
    print("开始异步任务")
    await asyncio.sleep(2)
    print("异步任务完成")
    return "done"

def run_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    loop_thread = threading.Thread(target=run_loop, args=(loop,), daemon=True)
    loop_thread.start()

    # 从非事件循环线程提交协程，返回 concurrent.futures.Future。
    future = asyncio.run_coroutine_threadsafe(async_task(), loop)

    try:
        print(future.result(timeout=5))
    finally:
        loop.call_soon_threadsafe(loop.stop)
        loop_thread.join()
        loop.close()

代码解析

创建事件循环：使用 asyncio.new_event_loop() 创建事件循环，并在后台线程中运行。
定义异步任务：async_task 是一个异步协程，模拟一个需要 2 秒的耗时操作。
调度异步任务：使用 asyncio.run_coroutine_threadsafe 将 async_task 调度到指定事件循环中。它返回的是 concurrent.futures.Future，可以通过 result() 获取结果或异常。
关闭事件循环：使用 loop.call_soon_threadsafe(loop.stop) 从其他线程安全地通知事件循环停止，最后关闭 loop。