企业官网建设流程全解析

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(), timeout=3) 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_pages=5): 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 开发的重要技能，希望本文能帮助你开启这段学习之旅。