Python使用Asyncio如何实现两个异步函数同时并发运行？

• 协作式调度特性：Asyncio的事件循环默认在单线程中运行，依赖任务主动让出控制权才会切换。你代码中funcA和funcb内的大循环都是纯同步CPU计算逻辑，运行时会独占线程，不会给其他任务执行机会。
• 调用逻辑阻塞：main函数中使用await funcA()的写法，会等待当前funcA完全执行完毕才会进入下一轮循环，连多个funcA都无法并发，更不会调度后台的funcb任务。

方案1：适配IO密集场景 #

如果你的真实业务逻辑是IO密集型（比如有网络请求、文件读写），只需要调整调度逻辑，在需要让出的位置插入await asyncio.sleep(0)主动让出控制权，同时修改main函数的并发逻辑即可：

import asyncio
import random

queue = []

async def funcA():
    r = random.randint(0, 1000)
    for i in range(10000000):
        # 按实际业务场景调整让出频率，避免长时间占用事件循环
        if i % 10000 == 0:
            await asyncio.sleep(0)
        pass
    print(r)
    queue.append(r)
    asyncio.create_task(funcb(r))

async def funcb(r):
    for i in range(100000000):
        if i % 10000 == 0:
            await asyncio.sleep(0)
        pass
    print(r, "XX")
    queue.pop()


async def main():
    # 批量创建所有funcA任务，不要逐个await阻塞
    tasks = [asyncio.create_task(funcA()) for _ in range(10)]
    await asyncio.gather(*tasks)
    # 等待所有funcb执行完毕再打印队列
    while queue:
        await asyncio.sleep(0.1)
    print(queue)


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

方案2：适配CPU密集场景 #

如果你的业务逻辑确实是CPU密集型，Asyncio单线程本身不适合这类场景，用asyncio.to_thread（Python3.9+可用，低版本可以用loop.run_in_executor）把计算部分扔到线程池运行，不会阻塞事件循环：

import asyncio
import random

queue = []

# 纯CPU计算逻辑抽为同步函数，扔到线程池运行
def cpu_bound_task(count):
    for i in range(count):
        pass

async def funcA():
    r = random.randint(0, 1000)
    await asyncio.to_thread(cpu_bound_task, 10000000)
    print(r)
    queue.append(r)
    asyncio.create_task(funcb(r))

async def funcb(r):
    await asyncio.to_thread(cpu_bound_task, 100000000)
    print(r, "XX")
    queue.pop()


async def main():
    tasks = [asyncio.create_task(funcA()) for _ in range(10)]
    await asyncio.gather(*tasks)
    while queue:
        await asyncio.sleep(0.1)
    print(queue)


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

如果是funcA无限运行的场景，只需要在funcA的无限循环里每隔一段逻辑插入一次await asyncio.sleep(0)（IO密集场景）或者把循环内的计算逻辑扔到线程池执行，就能保证funcb被正常调度。

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