如何在后台运行asyncio任务并实现3秒后定时取消？

核心问题 #

1. 主线程阻塞：原代码中asyncio.run(run_ws())是阻塞调用，run_ws会一直等待custom_coro完成，但custom_coro是无限循环，导致主线程永远无法执行到print('b')及后续代码。
2. 事件循环被卡死：custom_coro里使用了同步的time.sleep(1)，会直接阻塞整个异步事件循环，即使有其他任务也无法被调度执行。

修正后的代码 #

import asyncio
import uuid


async def custom_coro():
    while True:
        print('a')
        # 替换同步sleep为异步sleep，让出事件循环给其他任务
        await asyncio.sleep(1)

async def main():
    # 创建后台任务运行custom_coro
    task = asyncio.create_task(custom_coro())
    tasks = {str(uuid.uuid4()): task}
    
    # 打印一次'b'
    print('b')
    
    # 异步等待3秒，期间事件循环可正常调度custom_coro
    await asyncio.sleep(3)
    
    # 取消后台任务
    target_task = tasks[next(iter(tasks.keys()))]
    target_task.cancel()
    try:
        # 捕获任务取消异常，规范处理任务结束流程
        await target_task
    except asyncio.CancelledError:
        print("任务已取消")

if __name__ == '__main__':
    asyncio.run(main())

关键修改说明 #

• 使用asyncio.create_task将custom_coro转为后台任务，使其在事件循环中并发执行。
• 用await asyncio.sleep(1)替代time.sleep(1)，避免阻塞事件循环，让其他异步操作（打印'b'、等待3秒）能正常执行。
• 将所有逻辑整合到main协程中，通过asyncio.run(main())启动整个异步流程，避免主线程被提前阻塞。
• 取消任务后通过await捕获CancelledError，符合异步任务取消的规范处理方式。

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