如何创建线程安全的asyncio后台任务？获取asyncio.Task对象方法

要在主线程向独立线程的事件循环添加任务并获取asyncio.Task对象（而非concurrent.futures.Future），可以通过以下步骤实现：

核心思路 #

在目标事件循环所在线程中封装一个协程函数，由它调用asyncio.create_task创建任务并返回Task对象；主线程通过asyncio.run_coroutine_threadsafe调用这个协程，从返回的Future中提取Task对象。

代码示例 #

1. 启动独立线程的事件循环 #

import asyncio
from threading import Thread

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

# 创建并启动事件循环线程
target_loop = asyncio.new_event_loop()
loop_thread = Thread(target=run_event_loop, args=(target_loop,), daemon=True)
loop_thread.start()

2. 封装创建Task的协程 #

async def create_task_in_loop(coro, task_name):
    # 在目标事件循环内创建Task并返回
    task = asyncio.create_task(coro, name=task_name)
    return task

3. 主线程调用并获取Task对象 #

# 定义要执行的后台协程
async def background_task():
    await asyncio.sleep(3)
    print("后台任务执行完成")

# 主线程中触发Task创建
future = asyncio.run_coroutine_threadsafe(
    create_task_in_loop(background_task(), "my-background-task"),
    target_loop
)
# 获取返回的Task对象
task = future.result()

# 直接访问Task的name属性
print(f"任务名称: {task.get_name()}")  # 输出 "my-background-task"

4. 线程安全地操作Task #

注意：asyncio.Task对象并非线程安全，若要执行取消、获取结果等操作，需封装成协程再通过run_coroutine_threadsafe调用：

async def cancel_target_task(task):
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("任务已取消")

# 主线程发起取消操作
asyncio.run_coroutine_threadsafe(cancel_target_task(task), target_loop)

原理说明 #

• create_task_in_loop协程在目标事件循环的线程中执行，因此asyncio.create_task创建的Task属于该循环，能正常使用所有Task属性和方法。
• asyncio.run_coroutine_threadsafe负责跨线程传递协程执行请求，返回的Future对象的结果就是我们需要的Task实例。

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