如何修改Python代码使CPU密集型任务阻塞asyncio事件循环并互斥执行？

你之前的代码用了loop.run_in_executor把阻塞任务丢给线程池执行，线程池的抢占式调度会让两个任务交替跑，自然不会阻塞事件循环。要实现预期的阻塞效果，核心是让CPU密集型任务直接在asyncio的事件循环线程里执行——因为asyncio是单线程协作式调度，一旦有同步阻塞任务占住线程，整个事件循环就会被卡住，直到这个任务跑完才会处理其他任务。

修改后的代码如下：

import asyncio
import threading
from datetime import datetime

def log(msg):
    print(f"{datetime.now()} {msg}")

def blocking_task(task_name):
    num_iter = 100
    log(f"{task_name} started, thread_id: {threading.get_ident()}")
    for i in range(num_iter):
        log(f"{i}/{num_iter} executing {task_name} ...")
    log(f"{task_name} finished")

async def taskA():
    # 直接调用阻塞任务，不丢去线程池
    blocking_task("Blocking Task A")

async def taskB():
    # 同样直接调用阻塞任务
    blocking_task("Blocking Task B")

async def async_main():
    log(f"async_main thread_id: {threading.get_ident()}")
    
    task_a = asyncio.create_task(taskA())
    task_b = asyncio.create_task(taskB())

    await asyncio.gather(task_a, task_b)
    
    log("Async main finished")

if __name__ == "__main__":
    log(f"Main thread ID: {threading.get_ident()}")
    
    asyncio.run(async_main())
    
    log("Main thread finished")

修改说明 #

1. 移除了所有loop.run_in_executor调用：让blocking_task直接在事件循环线程中执行，不再借助线程池的多线程调度。
2. 简化了taskA和taskB：不需要再传入loop对象，直接调用同步阻塞函数即可。

效果验证 #

运行修改后的代码会看到：

• 先完整执行完Blocking Task A的所有迭代，直到它显示"finished"
• 之后才会开始执行Blocking Task B的迭代
  这就证明了两个任务互相阻塞，CPU密集型任务完全卡住了asyncio的事件循环。

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