Python asyncio 异步编程实战：从协程到任务调度 前言Python 的 asyncio 是编写并发代码的标准库，使用 async/await 语法实现单线程下的高并发 IO 操作。本文将从协程基础讲到实际应用，帮你掌握 Python 异步编程。一、同步 vs 异步1.1 同步代码的问题import requests import time def fetch_url(url): resp = requests.get(url) return resp.status_code start = time.time() for url in ["https://httpbin.org/delay/1"] * 10: fetch_url(url) print(f"同步耗时: {time.time() - start:.2f}s") # ~10s10 个请求串行执行，每个等待 1 秒，总共 10 秒。1.2 异步的优势import asyncio import aiohttp import time async def fetch_url(session, url): async with session.get(url) as resp: return resp.status async def main(): async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, "https://httpbin.org/delay/1") for _ in range(10)] results = await asyncio.gather(*tasks) print(results) start = time.time() asyncio.run(main()) print(f"异步耗时: {time.time() - start:.2f}s") # ~1s10 个请求并发执行，总共约 1 秒。二、协程基础2.1 定义和调用协程import asyncio async def hello(): print("Hello") await asyncio.sleep(1) print("World") asyncio.run(hello())2.2 await 的含义await 只能在 async 函数内使用，它会：暂停当前协程的执行将控制权交还给事件循环等待 awaitable 对象完成后恢复执行async def task_a(): print("A start") await asyncio.sleep(1) print("A end") async def task_b(): print("B start") await asyncio.sleep(1) print("B end") async def main(): await task_a() await task_b() asyncio.run(main())三、并发执行3.1 asyncio.gatherasync def main(): await asyncio.gather(task_a(), task_b())3.2 asyncio.create_taskasync def main(): task1 = asyncio.create_task(task_a()) task2 = asyncio.create_task(task_b()) print("doing other work") await task1 await task23.3 超时控制async def slow_operation(): await asyncio.sleep(10) return "done" async def main(): try: result = await asyncio.wait_for(slow_operation(), timeout=3.0) print(result) except asyncio.TimeoutError: print("操作超时!") asyncio.run(main())3.4 asyncio.as_completedasync def fetch(simulated_time, name): await asyncio.sleep(simulated_time) return f"{name}: {simulated_time}s" async def main(): tasks = [ fetch(3, "task1"), fetch(1, "task2"), fetch(2, "task3"), ] for coro in asyncio.as_completed(tasks): result = await coro print(result) asyncio.run(main()) # 输出顺序: task2, task3, task1四、实际应用：并发爬虫import asyncio import aiohttp async def fetch_page(session, url): async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp: text = await resp.text() return { "url": url, "status": resp.status, "length": len(text), } async def crawl(urls, concurrency=5): semaphore = asyncio.Semaphore(concurrency) async def limited_fetch(session, url): async with semaphore: return await fetch_page(session, url) async with aiohttp.ClientSession() as session: tasks = [limited_fetch(session, url) for url in urls] return await asyncio.gather(*tasks, return_exceptions=True) urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(20)] results = asyncio.run(crawl(urls, concurrency=5)) for r in results: if isinstance(r, Exception): print(f"Error: {r}") else: print(f"{r[\"status\"]} - {r[\"url\"]} ({r[\"length\"]} bytes)")五、异步上下文管理器class AsyncDBConnection: async def __aenter__(self): print("连接数据库...") await asyncio.sleep(0.5) return self async def __aexit__(self, exc_type, exc_val, exc_tb): print("关闭连接...") await asyncio.sleep(0.1) async def query(self, sql): await asyncio.sleep(0.1) return f"Result of: {sql}" async def main(): async with AsyncDBConnection() as db: result = await db.query("SELECT * FROM users") print(result) asyncio.run(main())六、常见陷阱6.1 在异步中调用同步阻塞代码# 错误：requests 是同步库，会阻塞事件循环 async def bad_fetch(url): return requests.get(url).text # 正确：使用 aiohttp async def good_fetch(url): async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.text()6.2 忘记 awaitasync def main(): asyncio.sleep(1) # 不会执行！返回 coroutine 对象 await asyncio.sleep(1) # 正确6.3 在异步代码中使用 time.sleepimport time async def main(): time.sleep(1) # 阻塞事件循环！ await asyncio.sleep(1) # 正确：让出控制权七、总结asyncio 的核心价值在于：在单线程内通过事件循环实现高并发 IO，避免了多线程的锁问题和上下文切换开销。记住三个关键点：IO 操作必须用异步库（aiohttp 而非 requests）、用 gather/create_task 实现并发、用 Semaphore 控制并发量。在高 IO 场景（爬虫、API 调用、数据库查询）中，asyncio 能带来数量级的性能提升。