Python 异步编程引言为什么需要异步编程在传统的同步编程中程序会按照代码顺序一行一行地执行如果遇到耗时操作如网络请求、文件读写、数据库查询整个程序会被阻塞住等待操作完成后才能继续执行。这就好比你在餐厅点餐服务员必须等你点完才能服务下一位顾客效率很低。异步编程的核心思想是在等待耗时操作时程序可以先去执行其他任务等耗时操作完成后再回来处理结果。这就像餐厅里多个服务员同时服务多位顾客大大提高了效率。Python 从 3.4 版本开始引入asyncio库提供了强大的异步编程支持。本文将带你从基础到高级全面掌握 Python 异步编程。## 基础概念### 协程Coroutine协程是异步编程的基本单位它类似于函数但可以被暂停和恢复。在 Python 中使用async def定义协程。pythonimport asyncio# 定义一个简单的协程async def say_hello(): print(Hello!) # 模拟耗时操作 await asyncio.sleep(1) print(World!)# 运行协程asyncio.run(say_hello())代码解释-async def定义了一个协程函数-await用于等待另一个协程完成同时让出控制权-asyncio.sleep(1)是一个非阻塞的等待操作-asyncio.run()是运行协程的入口点### await 关键字await是异步编程的核心关键字它有两个作用1. 等待一个可等待对象如协程、Future、Task完成2. 在等待期间让出 CPU 控制权让事件循环调度其他协程## 事件循环Event Loop事件循环是异步编程的调度器它负责- 注册和管理协程- 调度协程的执行- 处理 I/O 事件- 处理定时任务当我们调用asyncio.run()时它会自动创建并运行事件循环。pythonimport asyncioasync def task1(): print(Task 1 开始) await asyncio.sleep(2) print(Task 1 完成)async def task2(): print(Task 2 开始) await asyncio.sleep(1) print(Task 2 完成)async def main(): # 创建多个任务并发执行 task_a asyncio.create_task(task1()) task_b asyncio.create_task(task2()) # 等待所有任务完成 await task_a await task_b# 运行主协程asyncio.run(main())运行结果Task 1 开始Task 2 开始Task 2 完成Task 1 完成解释两个任务几乎同时开始task2 因为休眠时间较短而先完成task1 后完成。这体现了异步编程的并发特性。## 高级用法### 1. 并发执行多个任务使用asyncio.gather()可以同时运行多个协程并收集它们的返回结果。pythonimport asyncioimport timeasync def fetch_data(url, delay): 模拟网络请求 print(f开始请求: {url}) await asyncio.sleep(delay) # 模拟网络延迟 print(f完成请求: {url}) return f数据来自 {url}async def main(): # 同时发起三个请求 start_time time.time() results await asyncio.gather( fetch_data(https://api.example.com/1, 3), fetch_data(https://api.example.com/2, 2), fetch_data(https://api.example.com/3, 1) ) end_time time.time() print(f总耗时: {end_time - start_time:.2f} 秒) print(f结果: {results})asyncio.run(main())输出结果开始请求: https://api.example.com/1开始请求: https://api.example.com/2开始请求: https://api.example.com/3完成请求: https://api.example.com/3完成请求: https://api.example.com/2完成请求: https://api.example.com/1总耗时: 3.00 秒结果: [数据来自 https://api.example.com/1, 数据来自 https://api.example.com/2, 数据来自 https://api.example.com/3]关键点- 三个请求几乎同时开始- 总耗时由最长的请求决定3秒而不是总和6秒-gather()按传入顺序返回结果### 2. 超时控制使用asyncio.wait_for()设置协程的超时时间防止长时间阻塞。pythonimport asyncioasync def slow_operation(): 模拟慢操作 await asyncio.sleep(10) return 完成async def main(): try: # 设置超时时间为 3 秒 result await asyncio.wait_for(slow_operation(), timeout3) print(result) except asyncio.TimeoutError: print(操作超时)asyncio.run(main())### 3. 异步上下文管理器使用async with管理异步资源如数据库连接、文件操作等。pythonimport asyncioclass AsyncDatabase: 模拟异步数据库连接 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(0.5) # 模拟关闭连接 async def query(self, sql): 执行查询 print(f执行查询: {sql}) await asyncio.sleep(1) # 模拟查询延迟 return [{id: 1, name: Alice}]async def main(): async with AsyncDatabase() as db: # 使用数据库 result await db.query(SELECT * FROM users) print(f查询结果: {result})asyncio.run(main())输出打开数据库连接...执行查询: SELECT * FROM users查询结果: [{id: 1, name: Alice}]关闭数据库连接...### 4. 异步迭代器使用async for处理异步生成的数据流如分页加载、实时数据等。pythonimport asyncioclass AsyncPaginator: 模拟分页数据获取 def __init__(self, total_pages5): self.current_page 0 self.total_pages total_pages def __aiter__(self): return self async def __anext__(self): self.current_page 1 if self.current_page self.total_pages: raise StopAsyncIteration # 模拟每页数据加载延迟 await asyncio.sleep(0.5) return f第 {self.current_page} 页数据async def main(): print(开始加载分页数据...) async for page_data in AsyncPaginator(3): print(f获取到: {page_data})asyncio.run(main())### 5. 信号量Semaphore限制并发使用asyncio.Semaphore控制并发数量防止资源过载。pythonimport asyncioasync def api_call(semaphore, call_id): 模拟 API 调用 async with semaphore: print(f调用 {call_id} 开始) await asyncio.sleep(1) print(f调用 {call_id} 完成) return f结果_{call_id}async def main(): # 限制最多同时 3 个并发 semaphore asyncio.Semaphore(3) tasks [api_call(semaphore, i) for i in range(10)] results await asyncio.gather(*tasks) print(f所有结果: {results})asyncio.run(main())## 最佳实践与注意事项1.不要在协程中执行 CPU 密集型操作协程适合 I/O 密集型任务CPU 密集型操作会阻塞事件循环。对于 CPU 密集型任务使用concurrent.futures或asyncio.to_thread()。2.避免阻塞调用永远不要使用time.sleep()而要用asyncio.sleep()。类似地使用异步库替代同步库如aiohttp替代requests。3.合理使用asyncio.run()每个 Python 脚本只调用一次asyncio.run()。4.注意协程的创建时机async def定义的函数返回的是协程对象需要await才能执行。## 总结Python 异步编程通过协程、事件循环和await/async语法提供了一种高效处理 I/O 密集型任务的方式。本文从基础概念出发逐步深入到高级用法包括-核心概念协程、await、事件循环-并发控制gather()、超时管理、信号量-资源管理异步上下文管理器、异步迭代器-最佳实践避免阻塞、合理并发掌握这些知识后你就能编写出高效、可扩展的异步 Python 程序特别适合网络爬虫、Web 服务、实时数据处理等场景。异步编程是现代 Python 开发的重要技能希望本文能帮助你开启这段学习之旅。