如何在asyncio应用中利用空闲时间处理计算密集型任务？

这个问题在asyncio开发里太常见了——计算密集型的预计算任务很容易把单线程的事件循环堵得水泄不通，导致心跳这类延迟敏感的协程根本抢不到时间片，甚至因为超时终止。既然你不想拆成独立进程，那咱们可以从asyncio的调度机制和任务隔离入手，给你几个实用的单程序内解决方案：

1. 用线程池隔离计算密集型任务（最省心的方案） #

asyncio的事件循环是单线程的，计算密集型任务会阻塞整个循环，所以把这些预计算任务扔到线程池里是最直接的办法，不用改动太多计算逻辑。你可以用loop.run_in_executor把任务提交到自定义的线程池，同时控制线程池的大小，避免线程过多抢占CPU：

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 根据CPU核心数初始化小容量线程池，避免占用过多资源
executor = ThreadPoolExecutor(max_workers=2)

async def precompute_task(data):
    # 把计算任务扔到线程池，不阻塞主事件循环
    result = await asyncio.get_running_loop().run_in_executor(
        executor, heavy_computation, data
    )
    # 将结果写入缓冲区
    buffer.put(result)

def heavy_computation(data):
    # 你的计算密集型逻辑
    ...

这里的关键是控制线程池大小——别开太多线程，比如4核CPU开1-2个线程就够了，让计算任务“缓慢”运行，给心跳、延迟敏感模块留足CPU时间。如果想进一步降低计算任务的优先级，可以在提交前加await asyncio.sleep(0)，让事件循环先处理其他就绪任务。

2. 拆分计算任务为协程友好的小粒度 #

如果不想用线程池，那可以把大的计算任务拆成多个小步骤，每执行完一步就主动让出事件循环，给心跳等协程留运行机会：

async def precompute_task(data):
    chunk_size = 1000  # 根据计算逻辑调整拆分粒度
    total_result = []
    for i in range(0, len(data), chunk_size):
        chunk = data[i:i+chunk_size]
        # 执行一小段计算
        partial_result = heavy_computation_chunk(chunk)
        total_result.append(partial_result)
        # 主动让出事件循环，触发协程切换
        await asyncio.sleep(0)
    # 合并结果到缓冲区
    buffer.put(merge_results(total_result))

def heavy_computation_chunk(chunk):
    # 一小段独立的计算逻辑
    ...

这个方法的核心是避免长时间占用事件循环——await asyncio.sleep(0)会强制事件循环切换到其他就绪协程。拆分粒度要根据心跳频率调整：比如心跳每1秒一次，那每个计算chunk的执行时间要远小于1秒，保证心跳能及时触发。

3. 用优先级队列调度任务，保障高优先级协程执行 #

如果你的场景需要严格的优先级控制（比如心跳必须优先于预计算），可以用asyncio.PriorityQueue实现优先级任务调度器：

import asyncio

# 优先级队列：数值越小，优先级越高
priority_queue = asyncio.PriorityQueue()

async def task_worker():
    while True:
        priority, task_func, args = await priority_queue.get()
        try:
            await task_func(*args)
        finally:
            priority_queue.task_done()

# 心跳任务（高优先级，设为0）
async def heartbeat_task():
    while True:
        # 处理websocket心跳逻辑
        ...
        await asyncio.sleep(1)
        # 重新提交自己到优先级队列
        await priority_queue.put((0, heartbeat_task, ()))

# 提交预计算任务（低优先级，设为10）
async def submit_precompute(data):
    await priority_queue.put((10, precompute_task, (data,)))

async def precompute_task(data):
    # 预计算逻辑（建议结合线程池或任务拆分，避免阻塞）
    ...

程序启动时启动worker协程即可：

async def main():
    asyncio.create_task(task_worker())
    # 初始化心跳任务
    await priority_queue.put((0, heartbeat_task, ()))
    # 启动其他业务逻辑
    ...

这个方案能确保高优先级任务（比如心跳）总是先被执行，预计算任务只有在队列无高优先级任务时才会运行。注意：如果预计算本身是阻塞的，还是要结合线程池或任务拆分，不然即使优先级低，执行时仍会阻塞事件循环。

4. 利用事件循环空闲回调执行预计算 #

如果预计算任务不是紧急需求，只是需要“在系统空闲时”填充缓冲区，可以利用事件循环的空闲检查，只在无其他任务时执行预计算：

async def precompute_when_idle():
    while True:
        # 检查事件循环是否空闲（无待处理的IO/协程任务）
        loop = asyncio.get_running_loop()
        if loop._ready.empty():
            # 执行一次预计算
            result = await loop.run_in_executor(executor, heavy_computation)
            buffer.put(result)
        # 短暂等待后再次检查
        await asyncio.sleep(0.5)

这个方法适合后台填充缓冲区的场景，完全不会影响核心的延迟敏感任务。

总结 #

• 若计算逻辑不好修改，优先选线程池+控制线程数，简单有效；
• 想纯协程实现，就拆分计算粒度+主动让出事件循环；
• 需要严格优先级控制，用优先级队列调度；
• 非紧急预计算，用空闲回调。

所有方案的核心都是：避免让计算密集型任务长时间占用asyncio事件循环，给心跳、延迟敏感模块留出足够的执行时间。

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