Python 异步编程：Asyncio 入门指南

异步编程是处理I/O密集型任务的高效范式，而Asyncio是Python实现异步的核心库。本指南将带您掌握其核心概念与实践技巧。

一、异步编程基础

1. 同步 vs 异步

   • 同步：代码顺序执行，遇到I/O操作（如网络请求）时线程阻塞
   • 异步：I/O操作时挂起当前任务，切换到其他任务，实现并发

   效率对比（$n$个任务）： $$ \text{同步耗时} = \sum_{i=1}^{n} T_i $$ $$ \text{异步耗时} \approx \max(T_i) $$

2. 事件循环（Event Loop）

   • 异步程序的核心调度器
   • 管理任务队列，监控I/O事件
   • 工作流程：

     graph LR
     A[创建任务] --> B[事件循环]
     B --> C{可运行？}
     C -->|是| D[执行任务]
     C -->|否| E[挂起]
     D --> F[I/O完成]
     F --> B
     

二、Asyncio 核心组件

1. 协程（Coroutine）

   • 使用 async def 定义
   • 通过 await 挂起I/O操作

   async def fetch_data(url):
       print(f"开始请求 {url}")
       await asyncio.sleep(2)  # 模拟网络延迟
       return f"{url} 数据"
   

2. 任务（Task）

   • 协程的封装，可并发执行

   async def main():
       task1 = asyncio.create_task(fetch_data("api1"))
       task2 = asyncio.create_task(fetch_data("api2"))
       results = await asyncio.gather(task1, task2)
       print(results)  # 输出 ['api1 数据', 'api2 数据']
   

3. Future 对象

   • 代表未完成操作的容器
   • 常用方法：
     • .done() 检查状态
     • .result() 获取结果

三、实战模式

1. 并发控制

   async def batch_process(items):
       semaphore = asyncio.Semaphore(5)  # 最大并发数
       async def process(item):
           async with semaphore:
               return await process_item(item)
       return await asyncio.gather(*[process(i) for i in items])
   

2. 超时处理

   try:
       await asyncio.wait_for(fetch_data(), timeout=3.0)
   except asyncio.TimeoutError:
       print("请求超时")
   

3. 任务取消

   task = asyncio.create_task(long_running_job())
   await asyncio.sleep(1)
   task.cancel()
   try:
       await task
   except asyncio.CancelledError:
       print("任务已取消")
   

四、应用场景

1. Web服务：FastAPI/Quart异步路由
2. 爬虫系统：并发抓取数百页面
3. 实时通信：WebSocket消息处理
4. 微服务调用：聚合多个API响应

性能提升示例（$k$个请求）： $$ \text{同步耗时} \propto k \cdot T_{\text{latency}} $$ $$ \text{异步耗时} \propto T_{\text{latency}} $$

五、最佳实践

1. 避免在协程内使用阻塞调用（如 time.sleep()）
2. 使用 asyncio.run() 管理主入口
3. 优先选择异步库（如 aiohttp 替代 requests）
4. 监控任务状态：asyncio.all_tasks()

# 完整示例：并发获取网页标题
import aiohttp

async def get_title(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            html = await response.text()
            return html.split("<title>")[1].split("</title>")[0]

async def main():
    urls = ["https://python.org", "https://github.com"]
    titles = await asyncio.gather(*[get_title(u) for u in urls])
    print(titles)  # 输出 ['Welcome to Python.org', 'GitHub...']

asyncio.run(main())

掌握Asyncio可显著提升程序吞吐量，特别适合高并发场景。建议从简单任务开始实践，逐步构建复杂异步系统。