Python 异步编程深入：asyncio 的事件循环与协程调度

# Python 异步编程深入：asyncio 的事件循环与协程调度 ## 事件循环的本质 asyncio 的事件循环本质是一个单线程的 I/O 多路复用器（基于 epoll/kqueue），不断轮询就绪事件并执行对应的回调。 ```python import asyncio async def fetch_data(url, delay): print('开始请求: ' + url) await asyncio.sleep(delay) # 模拟网络 I/O print('完成请求: ' + url) return {'url': url, 'data': 'response'} ``` ## gather：并发执行多个协程 ```python async def main(): results = await asyncio.gather( fetch_data('/api/users', 1), fetch_data('/api/orders', 2), fetch_data('/api/products', 1.5), ) print('全部完成，共 {} 个结果'.format(len(results))) asyncio.run(main()) ``` 三个请求并发执行，总耗时约 2 秒（取决于最慢的那个）。 ## TaskGroup（Python 3.11+） ```python async def main(): async with asyncio.TaskGroup() as tg: task1 = tg.create_task(fetch_data('/api/users', 1)) task2 = tg.create_task(fetch_data('/api/orders', 2)) # TaskGroup 退出时所有任务已完成 print(task1.result(), task2.result()) ``` TaskGroup 的优势：任何一个任务异常时，会自动取消其他任务。 ## 信号量：控制并发数 ```python semaphore = asyncio.Semaphore(10) # 最多 10 个并发 async def limited_fetch(url): async with semaphore: return await fetch_data(url, 0.5) async def main(): urls = ['/api/item/{}'.format(i) for i in range(100)] tasks = [limited_fetch(url) for url in urls] results = await asyncio.gather(*tasks) print('完成 {} 个请求'.format(len(results))) ``` ## 常见陷阱 1. **忘记 await**：协程不 await 就不会执行 2. **阻塞调用**：`time.sleep()` 会阻塞事件循环，用 `asyncio.sleep()` 3. **CPU 密集型任务**：用 `loop.run_in_executor()` 放到线程池 ```python import concurrent.futures async def cpu_bound(): loop = asyncio.get_event_loop() with concurrent.futures.ProcessPoolExecutor() as pool: result = await loop.run_in_executor(pool, heavy_computation) return result ```