Python异步编程深度解析：从生成器到协程

异步编程是Python进阶路上的一道分水岭。很多人会用async/await但遇到复杂场景就不知道怎么办了。这篇文章从底层原理讲起帮你真正理解Python的异步编程。一、从同步到异步为什么需要异步先看一个典型的同步代码pythonimport time import requests def fetch_data(url): print(f开始请求: {url}) response requests.get(url) print(f请求完成: {url}) return response.json() def main(): urls [ https://api.example.com/user/1, https://api.example.com/user/2, https://api.example.com/user/3, ] start time.time() results [fetch_data(url) for url in urls] print(f总耗时: {time.time() - start:.2f}秒) # 如果每个请求耗时1秒总耗时约3秒同步代码的问题是程序在等待网络响应时CPU处于空闲状态。如果能利用这些等待时间去干别的事效率就能大幅提升。异步编程的核心思想就是遇到IO等待时让出CPU给其他任务执行。二、生成器异步的基础在讲协程之前必须先理解生成器。生成器是Python实现协程的基础。生成器的基本用法pythondef simple_generator(): print(开始执行) yield 1 print(继续执行) yield 2 print(结束执行) yield 3 gen simple_generator() print(next(gen)) # 输出: 开始执行 / 1 print(next(gen)) # 输出: 继续执行 / 2 print(next(gen)) # 输出: 结束执行 / 3 # print(next(gen)) # StopIteration生成器的核心特性使用yield暂停函数执行调用next()恢复执行函数的状态局部变量、程序计数器会被保留生成器可以接收数据pythondef echo_generator(): while True: received yield print(f收到: {received}) gen echo_generator() next(gen) # 启动生成器执行到yield gen.send(hello) # 发送数据 gen.send(world)send()方法可以向生成器发送数据这让生成器既可以产出数据也可以接收数据。这就是协程的雏形。从生成器到协程的演变python# Python 3.3 引入 yield from def sub_generator(): yield 1 yield 2 yield 3 def main_generator(): # yield from 会代理子生成器 yield from sub_generator() yield 4 for value in main_generator(): print(value) # 1, 2, 3, 4yield from让生成器可以委托给另一个生成器这为协程的嵌套调用奠定了基础。三、asyncio事件循环驱动一切asyncio是Python异步编程的核心库。它的工作原理可以简单理解为有一个事件循环Event Loop任务协程被提交到事件循环事件循环不断轮询执行就绪的任务遇到await等待时任务让出控制权等待的事件完成后任务被重新调度一个简单的事件循环实现为了理解原理我们手写一个简化版的事件循环pythonfrom collections import deque from types import GeneratorType class SimpleEventLoop: def __init__(self): self.tasks deque() def create_task(self, coro): 将协程加入任务队列 self.tasks.append(coro) def run_until_complete(self): 运行事件循环直到所有任务完成 while self.tasks: task self.tasks.popleft() try: # 驱动协程执行 result next(task) # 如果协程返回的是迭代器继续执行 if isinstance(result, GeneratorType): self.tasks.append(result) except StopIteration: pass def task(name, n): 一个简单的协程任务 for i in range(n): print(f{name}: {i}) yield # 让出控制权 # 使用 loop SimpleEventLoop() loop.create_task(task(任务A, 3)) loop.create_task(task(任务B, 2)) loop.run_until_complete() # 输出: 任务A: 0, 任务B: 0, 任务A: 1, 任务B: 1, 任务A: 2这个简化版展示了事件循环的核心机制任务交替执行遇到yield就切换。四、async/await语法糖背后的本质Python 3.5正式引入async/await语法它们本质上是生成器的语法糖。python# 使用async定义协程 async def hello(): return Hello # 调用协程不会执行返回一个协程对象 coro hello() print(coro) # coroutine object hello at 0x... # 需要在事件循环中运行 import asyncio result asyncio.run(hello()) print(result) # Helloawait相当于yield from的升级版pythonasync def fetch_data(): await asyncio.sleep(1) # 模拟IO等待 return data async def main(): # await会等待协程完成 result await fetch_data() print(result)实际执行流程await告诉事件循环我要等待一个操作完成当前协程让出控制权事件循环去执行其他任务等待的操作完成后事件循环恢复这个协程五、创建和运行异步任务基础用法pythonimport asyncio import time async def task(name, delay): print(f{name}: 开始) await asyncio.sleep(delay) print(f{name}: 完成) return f{name}的结果 async def main(): # 方式1await逐个执行串行 start time.time() result1 await task(任务1, 2) result2 await task(任务2, 1) print(f串行耗时: {time.time() - start:.2f}秒) # 方式2并发执行 start time.time() # 创建任务协程自动调度 t1 asyncio.create_task(task(任务3, 2)) t2 asyncio.create_task(task(任务4, 1)) # 等待所有任务完成 results await asyncio.gather(t1, t2) print(f并发耗时: {time.time() - start:.2f}秒) asyncio.run(main())并发控制限制同时执行的数量pythonimport asyncio import aiohttp async def fetch_url(session, url, semaphore): async with semaphore: # 限制并发数 async with session.get(url) as response: return await response.text() async def crawl_many(urls, max_concurrent10): semaphore asyncio.Semaphore(max_concurrent) async with aiohttp.ClientSession() as session: tasks [ fetch_url(session, url, semaphore) for url in urls ] results await asyncio.gather(*tasks) return results # 爬取100个页面同时最多10个请求 urls [fhttp://example.com/page/{i} for i in range(100)] results asyncio.run(crawl_many(urls, 10))超时控制pythonimport asyncio async def slow_operation(): await asyncio.sleep(10) return 完成 async def main(): try: # 方式1使用timeout result await asyncio.wait_for(slow_operation(), timeout5) except asyncio.TimeoutError: print(操作超时) # 方式2使用timeout上下文管理器 try: async with asyncio.timeout(5): result await slow_operation() except TimeoutError: print(操作超时) asyncio.run(main())六、异步上下文管理器和异步迭代器异步上下文管理器pythonimport asyncio import aiohttp class AsyncResource: async def __aenter__(self): print(获取资源) await asyncio.sleep(1) return self async def __aexit__(self, exc_type, exc_val, exc_tb): print(释放资源) await asyncio.sleep(1) async def use_resource(): async with AsyncResource() as resource: print(使用资源) asyncio.run(use_resource())python# 使用contextlib简化 from contextlib import asynccontextmanager asynccontextmanager async def get_db_connection(): conn await create_connection() try: yield conn finally: await conn.close() async def query_data(): async with get_db_connection() as conn: return await conn.execute(SELECT * FROM users)异步迭代器pythonimport asyncio class AsyncCounter: def __init__(self, max_count): self.max_count max_count self.count 0 def __aiter__(self): return self async def __anext__(self): self.count 1 if self.count self.max_count: raise StopAsyncIteration await asyncio.sleep(0.1) return self.count async def main(): async for num in AsyncCounter(10): print(num) asyncio.run(main())python# 使用async生成器Python 3.6 async def async_counter(max_count): for i in range(1, max_count 1): await asyncio.sleep(0.1) yield i async def main(): async for num in async_counter(10): print(num) asyncio.run(main())七、异步中的并发原语锁Lockpythonimport asyncio class Counter: def __init__(self): self.value 0 self.lock asyncio.Lock() async def increment(self): async with self.lock: # 临界区 current self.value await asyncio.sleep(0.01) # 模拟IO self.value current 1 async def worker(counter, n): for _ in range(n): await counter.increment() async def main(): counter Counter() # 创建10个并发任务每个增加100次 tasks [worker(counter, 100) for _ in range(10)] await asyncio.gather(*tasks) print(f最终值: {counter.value}) # 应该是1000 asyncio.run(main())事件Eventpythonimport asyncio async def waiter(event): print(等待事件...) await event.wait() print(事件已触发继续执行) async def setter(event): await asyncio.sleep(2) print(触发事件) event.set() async def main(): event asyncio.Event() await asyncio.gather( waiter(event), setter(event) ) asyncio.run(main())队列Queuepythonimport asyncio import random async def producer(queue, name): for i in range(5): item f{name}-{i} await queue.put(item) print(f生产者{name} 生产: {item}) await asyncio.sleep(random.random()) await queue.put(None) # 结束信号 async def consumer(queue, name): while True: item await queue.get() if item is None: break print(f消费者{name} 消费: {item}) await asyncio.sleep(random.random() * 0.5) queue.task_done() async def main(): queue asyncio.Queue(maxsize10) producers [producer(queue, fP{i}) for i in range(2)] consumers [consumer(queue, fC{i}) for i in range(3)] await asyncio.gather( *producers, *consumers ) asyncio.run(main())八、异步中的常见陷阱陷阱1忘记awaitpython# ❌ 错误 async def get_data(): return data async def main(): result get_data() # 没有awaitresult是一个协程对象 print(result) # coroutine object get_data at 0x... # ✅ 正确 result await get_data()陷阱2阻塞操作python# ❌ 错误使用阻塞的time.sleep async def bad_example(): time.sleep(1) # 会阻塞整个事件循环 # ✅ 正确使用异步版本 async def good_example(): await asyncio.sleep(1) # ❌ 错误使用阻塞的requests async def bad_request(): response requests.get(http://example.com) # 阻塞 # ✅ 正确使用aiohttp async def good_request(): async with aiohttp.ClientSession() as session: async with session.get(http://example.com) as resp: return await resp.text()陷阱3创建任务但未等待python# ❌ 问题创建了任务但没有等待 async def main(): asyncio.create_task(long_task()) # 任务可能还没执行完就结束了 print(主函数结束) # ✅ 解决等待所有任务完成 async def main(): task asyncio.create_task(long_task()) await task # 等待任务完成 # 或者使用asyncio.gather # await asyncio.gather(task)陷阱4在同步函数中调用异步函数python# ❌ 错误不能在同步函数中用await def sync_function(): result await async_function() # SyntaxError # ✅ 解决1将整个函数改为异步 async def async_wrapper(): return await async_function() # ✅ 解决2使用asyncio.run() def sync_function(): result asyncio.run(async_function()) return result九、实战异步爬虫下面是一个完整的异步爬虫示例综合运用了上述知识pythonimport asyncio import aiohttp from aiohttp import ClientTimeout, TCPConnector import json from typing import List, Dict import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class AsyncCrawler: def __init__(self, max_concurrent20, timeout30): self.max_concurrent max_concurrent self.timeout ClientTimeout(totaltimeout) self.semaphore asyncio.Semaphore(max_concurrent) self.session None async def __aenter__(self): connector TCPConnector(limitself.max_concurrent) self.session aiohttp.ClientSession( connectorconnector, timeoutself.timeout, headers{ User-Agent: Mozilla/5.0 (compatible; AsyncCrawler/1.0) } ) return self async def __aexit__(self, *args): await self.session.close() async def fetch(self, url: str, retry: int 3) - Dict: 抓取单个URL支持重试 for attempt in range(retry): try: async with self.semaphore: logger.info(f抓取: {url}) async with self.session.get(url) as response: if response.status ! 200: raise Exception(f状态码: {response.status}) # 根据Content-Type决定解析方式 content_type response.headers.get(Content-Type, ) if json in content_type: return await response.json() else: return {content: await response.text(), url: url} except asyncio.TimeoutError: logger.warning(f{url} 超时重试 {attempt1}/{retry}) except Exception as e: logger.warning(f{url} 错误: {e}重试 {attempt1}/{retry}) await asyncio.sleep(1) return {error: True, url: url} async def crawl(self, urls: List[str]) - List[Dict]: 批量抓取 tasks [self.fetch(url) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理异常结果 final_results [] for r in results: if isinstance(r, Exception): final_results.append({error: True, exception: str(r)}) else: final_results.append(r) return final_results async def main(): urls [ https://api.github.com/users/octocat, https://api.github.com/users/defunkt, https://api.github.com/users/github, https://httpbin.org/delay/2, # 故意延迟 https://httpbin.org/status/404, # 会出错 ] async with AsyncCrawler(max_concurrent3, timeout10) as crawler: results await crawler.crawl(urls) # 统计结果 success [r for r in results if not r.get(error)] failed [r for r in results if r.get(error)] print(f成功: {len(success)}失败: {len(failed)}) for result in results: if result.get(error): print(f❌ {result.get(url, unknown)}: {result.get(exception, unknown error)}) else: print(f✅ {result.get(url, unknown)}) if __name__ __main__: asyncio.run(main())十、异步性能对比来一个直观的对比pythonimport time import asyncio import requests import aiohttp # 同步版本 def sync_demo(): urls [https://httpbin.org/delay/1] * 10 start time.time() for url in urls: requests.get(url) return time.time() - start # 异步版本 async def async_demo(): urls [https://httpbin.org/delay/1] * 10 start time.time() async with aiohttp.ClientSession() as session: tasks [session.get(url) for url in urls] responses await asyncio.gather(*tasks) for resp in responses: await resp.text() return time.time() - start # 运行对比 print(f同步耗时: {sync_demo():.2f}秒) # 约10秒 print(f异步耗时: {asyncio.run(async_demo()):.2f}秒) # 约1秒写在最后异步编程是Python性能优化的利器但也是需要认真学习的知识点。学习建议先理解事件循环的概念从简单的asyncio.sleep()开始试验逐步引入网络请求、数据库操作遇到问题多调试理解协程的生命周期阅读优秀框架的源码如FastAPI、aiohttp记住异步不是万能药。对于CPU密集型任务异步帮不上忙对于简单的IO操作异步可能带来不必要的复杂度。选择合适的工具解决合适的问题。