Python asyncio如何在CPU密集协程运行期间执行周期保活任务

• asyncio的事件循环是单线程运行的，你当前写的cpu_intensive协程里是纯CPU计算的死循环，运行期间不会主动交出事件循环控制权，会直接阻塞整个事件循环，导致keep_alive协程永远得不到调度。
• 现有keep_alive只执行一次，没有周期性循环执行的逻辑，主协程也没有等待任务完成的逻辑，程序会直接提前退出。

核心思路是把CPU密集型任务放到独立线程运行，避免阻塞事件循环；保活任务循环执行，直到CPU任务完成后主动取消，具体实现代码如下：

import asyncio
import time

# CPU密集任务改为普通同步函数，后续放到独立线程运行
def cpu_intensive():
    print("cpu_intensive for 3 seconds")
    delay = 3
    close_time = time.time() + delay
    while True:
        if time.time() > close_time:
            break

async def keep_alive():
    # 加死循环实现周期性执行
    while True:
        print("keep alive for 1 second")
        await asyncio.sleep(1)

async def main():
    print(f"Started at {time.strftime('%X')}")
    # 将CPU密集任务扔到线程池执行，不阻塞事件循环
    cpu_task = asyncio.to_thread(cpu_intensive)
    # 如果你用的是Python3.9以下版本，把上面一行替换为下面的代码即可：
    # cpu_task = asyncio.get_event_loop().run_in_executor(None, cpu_intensive)
    keep_alive_task = asyncio.create_task(keep_alive())
    
    # 等待CPU密集任务执行完成
    await cpu_task
    # CPU任务结束后取消保活任务
    keep_alive_task.cancel()
    # 捕获取消异常避免报错
    try:
        await keep_alive_task
    except asyncio.CancelledError:
        pass
    
    print(f"Finished at {time.strftime('%X')}")

asyncio.run(main())

运行上述代码就能完全匹配你给出的预期输出，3秒运行周期内刚好打印3次保活日志。如果你的CPU密集任务逻辑可以拆分，也可以在计算间隙加await asyncio.sleep(0)主动交出控制权，但这种方案不如线程池稳定，计算量大时优先推荐线程池方案。

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