Python asyncio：同步CPU密集型任务与异步协程的数据交互实现问询

这个场景完全可行，核心思路是将CPU密集型的同步任务放到线程池/进程池中执行，彻底避免阻塞asyncio的事件循环，同时利用asyncio原生机制实现结果传递。以下是两种可靠的实现方案：

方案1：直接用asyncio.run_in_executor（最简洁） #

run_in_executor会把同步任务提交到指定的executor（默认是线程池），返回一个可await的Future对象。异步任务只需await这个Future，就能在同步任务完成后自动获取结果，全程不阻塞事件循环。

示例代码：

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

# CPU密集型同步任务
def cpu_intensive_task(input_data):
    time.sleep(5)  # 模拟长时间计算
    return input_data * 2

# 异步任务
async def async_task():
    print("异步任务开始执行")
    # 提交同步任务到线程池
    executor = ThreadPoolExecutor()
    result = await asyncio.get_running_loop().run_in_executor(
        executor, cpu_intensive_task, 10
    )
    print(f"异步任务收到同步任务结果: {result}")
    # 继续执行其他异步操作
    await asyncio.sleep(1)
    print("异步任务完成")

async def main():
    await async_task()

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

如果你的CPU密集任务受GIL限制（比如纯Python计算），可以替换为ProcessPoolExecutor，用法完全一致，能利用多CPU核心提升效率。

方案2：用asyncio.Queue传递结果（适合复杂协作场景） #

如果需要更灵活的任务调度（比如多个同步任务向同一个异步任务传递结果），可以用asyncio.Queue，但注意同步线程不能直接调用队列的异步方法，必须通过事件循环的call_soon_threadsafe方法来操作队列。

示例代码：

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

def cpu_intensive_task(input_data, queue, loop):
    time.sleep(5)
    result = input_data * 2
    # 同步线程中必须用call_soon_threadsafe调用队列的put方法
    loop.call_soon_threadsafe(queue.put_nowait, result)

async def async_task():
    print("异步任务开始执行")
    queue = asyncio.Queue()
    loop = asyncio.get_running_loop()
    executor = ThreadPoolExecutor()
    # 提交同步任务到线程池
    loop.run_in_executor(
        executor, cpu_intensive_task, 10, queue, loop
    )
    # 异步等待队列结果，不阻塞事件循环
    result = await queue.get()
    print(f"异步任务收到同步任务结果: {result}")
    await asyncio.sleep(1)
    print("异步任务完成")

async def main():
    await async_task()

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

为什么之前的队列方案失效？ #

• 用asyncio.Queue时直接调用put()：这是异步方法，在同步线程中无法正确执行，导致队列从未收到数据，get()一直等待。
• 用普通queue.Queue的get()：这是阻塞方法，在async协程中调用会直接卡住整个事件循环，所有异步任务都无法执行。

核心注意点 #

1. 绝对不能在asyncio协程中直接运行CPU密集型同步任务，会彻底阻塞事件循环。
2. 所有跨线程操作asyncio对象（如Future、Queue）时，必须通过loop.call_soon_threadsafe方法，避免线程安全问题。

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