为什么异步编程在 Python 中如此重要

写爬虫的人最先体会到同步 IO 的痛苦：依次请求二十个接口，每个等两秒，光网络等待就耗掉四十秒；而其中 CPU 几乎一直在空转，等待网卡把数据送回来。Python 因为全局解释器锁（GIL），多线程在 CPU 密集场景下发挥有限，但在 IO 密集场景下，等待时间才是瓶颈——这时异步编程就成了最自然的选择。

异步编程的核心思想是：在等待 IO 的时候，不要干等着，而是把控制权交出去，去做别的事，等数据到了再回来接着干。它不是多线程，也不是多进程，而是在单个线程内通过事件循环调度大量并发任务，避免了线程切换的开销。

asyncio 不是银弹：它最擅长的是 IO 密集型任务（网络请求、文件读写、数据库查询），CPU 密集型任务用它反而会变慢，因为事件循环本身被算满了。
本文核心知识点
  • 协程用 async def 定义的可暂停函数，遇到 await 让出控制权。
  • 事件循环负责调度所有协程，在等待时切换到其他可运行任务。
  • Task协程被包装成任务后才会被事件循环调度执行。
  • 并发控制信号量限制同时执行的任务数，防止压垮目标服务。
Python 异步编程示意：蓝色渐变背景上多条平行光流汇入中间事件循环核心，周围漂浮半透明任务方块并由发光线条连接
事件循环是大脑：成千上万个协程像流水线一样被它调度，等待时让出，就绪时继续。

协程与事件循环：asyncio 的核心模型

理解 asyncio 只需要抓住两个角色：协程（coroutine）和事件循环（event loop）。协程是用 async def 定义的函数，它和普通函数最大的区别是——它可以在 await 处暂停自己，把控制权还给事件循环，等等待的结果就绪后再恢复执行。

事件循环就像一个不停转的调度器：它手里维护着一堆就绪的任务，挨个执行；遇到任务说"我要等网络"，就把它挂起，转去执行下一个就绪任务；等网络数据到了，再把这个任务标记为就绪，下一轮接着跑。整个过程在一个线程里完成，没有线程切换，没有锁竞争。

概念作用类比
协程函数用 async def 定义，调用后返回协程对象一道工序
事件循环调度所有任务，管理就绪与等待状态车间主任
await在等待点暂停当前协程，交出控制权工序里的等待信号
Task协程被包装成任务后加入事件循环队列派工单
Future表示一个还未完成的异步结果待领取的快递

Python 3.7 之后，最常用的入口是 asyncio.run()：它创建事件循环、运行指定协程、关闭循环，一步到位。日常写异步脚本，入口永远是 asyncio.run(main())，不要手动去拿事件循环对象。

async / await 语法与任务创建

先看一个最基础的例子，理解 async 和 await 各自做了什么：

import asyncio
import time

async def say_after(delay: float, what: str) -> None:
    await asyncio.sleep(delay)
    print(what)

async def main():
    # 串行：总共约 3 秒
    t0 = time.perf_counter()
    await say_after(1, "你好")
    await say_after(2, "世界")
    print(f"串行耗时：{time.perf_counter() - t0:.2f} 秒")

asyncio.run(main())

上面这段代码虽然用了 async，但两个 await 是依次等待的，总耗时约 3 秒，和同步代码没区别。要真正并发，需要把协程包装成 Task，让事件循环同时调度它们：

async def main():
    t0 = time.perf_counter()
    # 包装成任务，立刻开始执行
    task1 = asyncio.create_task(say_after(1, "你好"))
    task2 = asyncio.create_task(say_after(2, "世界"))

    # 等待两个任务都完成
    await task1
    await task2
    print(f"并发耗时：{time.perf_counter() - t0:.2f} 秒")   # 约 2 秒

关键区别在 create_task：调用它的瞬间，协程就被提交到事件循环并开始运行，而不是等到 await 才开始。两个任务几乎同时启动，各自 sleep，谁先睡完谁先打印，总耗时取决于最长的那个，而不是两者之和。

  • 裸调用协程不会执行：say_after(1, "hi") 只是创建了协程对象，必须被 await 或包装成 Task 才会跑。
  • await 后面必须是可等待对象：协程、Task、Future 都可以，普通函数不行。
  • asyncio.sleep 不等于 time.sleep：前者是异步等待，不阻塞事件循环；后者会把整个循环卡死。

并发控制：gather、wait 与信号量

当你要并发执行成百上千个任务时，手工 create_task 再逐个 await 就太啰嗦了。asyncio 提供了两个高层接口来批量管理任务：asyncio.gather() 和 asyncio.wait()。

async def fetch_url(url: str) -> str:
    # 模拟一次网络请求
    await asyncio.sleep(0.5)
    return f"{url} -> 200 OK"

async def main():
    urls = [f"https://api.example.com/item/{i}" for i in range(20)]

    # gather：把所有协程打包并发执行，结果按顺序返回
    tasks = [fetch_url(u) for u in urls]
    results = await asyncio.gather(*tasks)
    for r in results:
        print(r)

gather 会把传入的所有协程并发跑起来，等全部完成后按原始顺序收集结果。如果某个任务抛异常，默认会立即把异常抛出来，其他任务仍在后台跑。想要"即使部分失败也拿到成功的结果"，加 return_exceptions=True。

但这里有个大坑：二十个请求同时打出去，目标服务器可能直接限流甚至封你 IP。真实场景里必须控制并发数，这就是信号量（Semaphore）的用途：

async def fetch_with_sem(sem: asyncio.Semaphore, url: str) -> str:
    async with sem:           # 同时最多 N 个任务进入这里
        return await fetch_url(url)

async def main():
    sem = asyncio.Semaphore(5)   # 限制并发为 5
    urls = [f"https://api.example.com/item/{i}" for i in range(100)]
    tasks = [fetch_with_sem(sem, u) for u in urls]
    await asyncio.gather(*tasks)

信号量就像停车场闸口：总共只有 5 个车位，满了后面的任务就在闸口外等，出来一个进一个。这是写并发爬虫和批量接口调用时的标配，既快又不会把服务打挂。

异步网络请求与并发抓取实战

asyncio 本身只提供事件循环和同步原语，真正发 HTTP 请求需要配合异步 HTTP 客户端。目前最主流的选择是 aiohttp，它和 requests 的思路类似，但所有 IO 方法都是协程。

import aiohttp
import asyncio

async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        data = await resp.json()
        return {"url": url, "status": resp.status, "title": data.get("title", "")}

async def main():
    urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 21)]
    sem = asyncio.Semaphore(5)

    async with aiohttp.ClientSession() as session:
        async def bounded(u):
            async with sem:
                return await fetch(session, u)
        results = await asyncio.gather(*[bounded(u) for u in urls])

    for r in results:
        print(r["status"], r["title"][:30])

asyncio.run(main())

这段代码做了三件正确的事：复用同一个 ClientSession（它内部维护连接池，每次新建 session 会浪费 TCP 握手）、用信号量把并发压到 5、用 gather 收集全部结果。二十个请求按 5 个一批跑完，总耗时大约是四批乘以单批延迟，远快于逐个请求。

几个工程上的经验：超时必须显式设置，否则某个接口卡住会拖死整个任务；失败要重试，用 tenacity 之类的库包一层指数退避；不要在协程里写同步阻塞代码，比如 requests 调用、time.sleep、大段 CPU 计算——它们会卡住整个事件循环，让所有并发任务一起停摆。