Resources 实施手册

# 异步 Python 模式实现手册 本文件包含技能引用的详细模式、检查清单和代码示例。 ## 核心概念 ### 1. 事件循环 (Event Loop) 事件循环是 asyncio 的核心，负责管理和调度异步任务。 **关键特性：** - 单线程协作式多任务处理 - 调度协程执行 - 处理 I/O 操作而不阻塞 - 管理回调和 Future 对象 ### 2. 协程 (Coroutines) 使用 `async def` 定义的函数，可以暂停和恢复。 **语法：** python async def my_coroutine(): result = await some_async_operation() return result ### 3. 任务 (Tasks) 在事件循环上并发运行的已调度协程。 ### 4. Future 对象 表示异步操作最终结果的底层对象。 ### 5. 异步上下文管理器 支持 `async with` 以进行正确清理的资源。 ### 6. 异步迭代器 支持 `async for` 以迭代异步数据源的对象。 ## 快速开始 python import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Python 3.7+ asyncio.run(main()) ## 基础模式 ### 模式 1: 基础 Async/Await python import asyncio async def fetch_data(url: str) -> dict: """异步从 URL 获取数据。""" await asyncio.sleep(1) # 模拟 I/O return {"url": url, "data": "result"} async def main(): result = await fetch_data("https://api.example.com") print(result) asyncio.run(main()) ### 模式 2: 使用 gather() 进行并发执行 python import asyncio from typing import List async def fetch_user(user_id: int) -> dict: """获取用户数据。""" await asyncio.sleep(0.5) return {"id": user_id, "name": f"User {user_id}"} async def fetch_all_users(user_ids: List[int]) -> List[dict]: """并发获取多个用户。""" tasks = [fetch_user(uid) for uid in user_ids] results = await asyncio.gather(*tasks) return results async def main(): user_ids = [1, 2, 3, 4, 5] users = await fetch_all_users(user_ids) print(f"已获取 {len(users)} 个用户") asyncio.run(main()) ### 模式 3: 任务创建与管理 python import asyncio async def background_task(name: str, delay: int): """长时间运行的后台任务。""" print(f"{name} 已启动") await asyncio.sleep(delay) print(f"{name} 已完成") return f"来自 {name} 的结果" async def main(): # 创建任务 task1 = as