Python async/await 协程：CPU密集型任务的陷阱与解决方案

import asyncio
import time

async def long_function():
    """一个纯粹的CPU密集型任务，不涉及任何I/O或异步操作。"""
    print(f"Task {asyncio.current_task().get_name()}: long_function started...")
    for _ in range(50_000_000): # 大量循环，模拟耗时计算
        pass
    print(f"Task {asyncio.current_task().get_name()}: long_function finished.")

async def count_blocking():
    """包含阻塞性CPU任务的协程。"""
    for x in range(3):
        print(f"Count {x} in {asyncio.current_task().get_name()}")
        await long_function() # 等待一个CPU密集型任务

async def main_blocking():
    """主协程，启动两个阻塞性计数任务。"""
    task1 = asyncio.create_task(count_blocking(), name="Count-A")
    task2 = asyncio.create_task(count_blocking(), name="Count-B")
    await asyncio.gather(task1, task2)

if __name__ == "__main__":
    start_time = time.perf_counter()
    asyncio.run(main_blocking())
    end_time = time.perf_counter()
    print(f"\n总执行时间 (阻塞): {end_time - start_time:.2f} 秒")

Count 0 in Count-A
Task Count-A: long_function started...
Task Count-A: long_function finished.
Count 1 in Count-A
Task Count-A: long_function started...
Task Count-A: long_function finished.
Count 2 in Count-A
Task Count-A: long_function started...
Task Count-A: long_function finished.
Count 0 in Count-B
Task Count-B: long_function started...
Task Count-B: long_function finished.
...

import asyncio
import time

async def long_function_cooperative():
    """一个协作式的CPU密集型任务，周期性地交出控制权。"""
    task_name = asyncio.current_task().get_name()
    # print(f"Task {task_name}: long_function_cooperative started...")
    for i in range(50_000_000):
        # 每隔一定次数的循环，显式地交出控制权
        if i % 10_000_000 == 0 and i != 0:
            # print(f"Task {task_name}: Yielding at iteration {i}")
            await asyncio.sleep(0) # 关键：交出控制权
    print(f"Task {task_name}: long_function_cooperative finished.")

async def count_cooperative():
    """包含协作性CPU任务的协程。"""
    for x in range(3):
        print(f"Count {x} in {asyncio.current_task().get_name()}")
        await long_function_cooperative() # 等待一个协作性CPU任务

async def main_cooperative():
    """主协程，启动两个协作性计数任务。"""
    task1 = asyncio.create_task(count_cooperative(), name="Count-X")
    task2 = asyncio.create_task(count_cooperative(), name="Count-Y")
    await asyncio.gather(task1, task2)

if __name__ == "__main__":
    start_time = time.perf_counter()
    asyncio.run(main_cooperative())
    end_time = time.perf_counter()
    print(f"\n总执行时间 (协作): {end_time - start_time:.2f} 秒")

Count 0 in Count-X
Count 0 in Count-Y
Count 1 in Count-X
Count 1 in Count-Y
Count 2 in Count-X
Count 2 in Count-Y
Task Count-X: long_function_cooperative finished.
Task Count-Y: long_function_cooperative finished.
...

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor

def blocking_cpu_task(task_id, iterations):
    """一个阻塞的CPU密集型函数，适合在进程池中运行。"""
    print(f"Process {task_id}: Starting CPU-bound task with {iterations} iterations...")
    result = 0
    for i in range(iterations):
        result += i # 执行一些计算
    print(f"Process {task_id}: Finished CPU-bound task. Result: {result % 1000}")
    return f"Task {task_id} completed."

async def run_cpu_tasks_with_pool():
    """使用ProcessPoolExecutor异步运行CPU密集型任务。"""
    # 使用ProcessPoolExecutor创建进程池，max_workers=None表示使用CPU核心数
    with ProcessPoolExecutor(max_workers=2) as executor:
        loop = asyncio.get_running_loop()

        # 将CPU密集型任务提交到进程池，并等待其完成
        tasks = [
            loop.run_in_executor(executor, blocking_cpu_task, "Alpha", 50_000_000),
            loop.run_in_executor(executor, blocking_cpu_task, "Beta", 50_000_000)
        ]

        results = await asyncio.gather(*tasks)
        print("\n所有CPU密集型任务通过ProcessPoolExecutor完成:")
        for res in results:
            print(res)

if __name__ == "__main__":
    start_time = time.perf_counter()
    asyncio.run(run_cpu_tasks_with_pool())
    end_time = time.perf_counter()
    print(f"\n总执行时间 (ProcessPoolExecutor): {end_time - start_time:.2f} 秒")