4. 异步IO与协程：asyncio核心概念、async/await语法、事件循环与Task管理

说到高并发，很多人第一反应就是多线程。但说实话，线程切换的开销比你想象中大得多。我早年做一个爬虫项目，开了200个线程，结果CPU全耗在上下文切换上了，实际干活的时间不到一半。

后来我转向了异步IO。它的核心思路很简单：等IO的时候别闲着。你想想看，网络请求发出去后，大部分时间都在等响应。这段时间CPU完全可以去做别的事。这就是异步编程的底层逻辑。

4.1 从同步到异步：一个直观的对比

先看一个最简单的例子。假设我们要请求3个网页：

# 同步方式
import time
import requests

def fetch(url):
    print(f"开始请求: {url}")
    resp = requests.get(url)
    print(f"完成: {url}, 长度: {len(resp.text)}")
    return resp

start = time.time()
fetch("https://httpbin.org/delay/1")
fetch("https://httpbin.org/delay/1")
fetch("https://httpbin.org/delay/1")
print(f"同步耗时: {time.time() - start:.2f}s")
# 输出: 同步耗时: 3.0x s

每个请求等1秒，3个串行就是3秒。换成异步呢？

import asyncio
import aiohttp

async def fetch_async(url):
    print(f"开始请求: {url}")
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            text = await resp.text()
            print(f"完成: {url}, 长度: {len(text)}")
            return text

async def main():
    tasks = [
        fetch_async("https://httpbin.org/delay/1"),
        fetch_async("https://httpbin.org/delay/1"),
        fetch_async("https://httpbin.org/delay/1"),
    ]
    await asyncio.gather(*tasks)

start = time.time()
asyncio.run(main())
print(f"异步耗时: {time.time() - start:.2f}s")
# 输出: 异步耗时: 1.0x s

3个请求几乎同时完成，总耗时只有1秒多一点。这就是异步IO的魅力——并发等待，而不是并发执行。

4.2 async/await：协程的语法糖

Python 3.5引入了async和await关键字。说白了，它们就是协程的语法糖。但理解它们背后的机制，比记住语法更重要。

关键字	作用	类比理解
async def	定义一个协程函数	就像定义一个"可暂停"的函数
await	等待一个协程完成	就像说"我先等会儿，你先干别的"
async with	异步上下文管理器	用于管理异步资源的生命周期
async for	异步迭代器	用于处理异步生成的数据流

4.3 事件循环：异步的调度中心

事件循环是asyncio的心脏。它负责：

• 注册待执行的协程
• 调度协程的执行顺序
• 处理IO事件的回调
• 管理定时任务

我画了一张图，帮你理解事件循环的工作流程：

事件循环的工作机制，其实就是一个永不停歇的调度器。它从就绪队列里取出协程执行，遇到await就把协程挂起到等待队列，等IO完成后再移回就绪队列。如此循环往复。

4.4 Task管理：协程的"容器"

Task是asyncio中非常重要的概念。你可以把它理解成协程的"容器"——它包装了一个协程，并跟踪它的执行状态。

import asyncio

async def worker(name, delay):
    print(f"{name} 开始工作")
    await asyncio.sleep(delay)
    print(f"{name} 完成工作")
    return f"{name} 的结果"

async def main():
    # 创建Task
    task1 = asyncio.create_task(worker("任务A", 2))
    task2 = asyncio.create_task(worker("任务B", 1))
    
    print("Task已创建，但还没执行")
    
    # 等待所有Task完成
    results = await asyncio.gather(task1, task2)
    print(f"所有结果: {results}")
    
    # 检查Task状态
    print(f"task1完成: {task1.done()}")
    print(f"task2完成: {task2.done()}")

asyncio.run(main())

4.5 常用API速查

API	作用	使用场景
asyncio.run()	启动事件循环，执行入口协程	程序入口，Python 3.7+推荐
asyncio.create_task()	将协程包装为Task并调度	并发执行多个协程
asyncio.gather()	并发等待多个协程完成	批量执行独立任务
asyncio.wait()	更灵活地等待协程集合	需要超时或部分结果时
asyncio.sleep()	异步休眠，不阻塞事件循环	模拟IO等待、定时任务
asyncio.Queue()	异步队列，支持协程间通信	生产者-消费者模式

4.6 实战：超时控制与取消Task

实际项目中，你一定会遇到"某个请求卡住了"的情况。没有超时控制的异步代码，就是一颗定时炸弹。

import asyncio

async def slow_operation():
    print("开始慢操作")
    await asyncio.sleep(10)  # 模拟一个很慢的IO
    print("慢操作完成")
    return "结果"

async def main():
    try:
        # 设置3秒超时
        result = await asyncio.wait_for(slow_operation(), timeout=3)
        print(f"得到结果: {result}")
    except asyncio.TimeoutError:
        print("操作超时了！")
    
    # 手动取消Task
    task = asyncio.create_task(slow_operation())
    await asyncio.sleep(1)
    task.cancel()  # 取消任务
    
    try:
        await task
    except asyncio.CancelledError:
        print("Task被取消了")

asyncio.run(main())

4.7 避坑指南

写异步代码这几年，我总结了几条血泪教训：

• 不要在协程里用time.sleep()——它会阻塞整个事件循环。用asyncio.sleep()代替。
• 不要混用同步和异步库——比如在协程里用requests.get()，它会阻塞事件循环。用aiohttp代替。
• 注意Task的生命周期——创建Task后要确保它被await或gather，否则可能"泄漏"。
• 异常处理要到位——Task内部的异常不会自动传播，需要用task.exception()或gather(return_exceptions=True)来捕获。