Python异步编程完全指南：asyncio从入门到实战

为什么 Python 需要异步编程

在 I/O 密集型场景下，传统的同步代码会因为等待网络或磁盘响应而白白浪费大量时间。假设你需要请求 10 个接口，每个耗时 0.5 秒，同步写法需要 5 秒，而异步写法可能只需 0.5 秒多一点。这就是异步编程的价值：在等待 I/O 时不阻塞线程，让事件循环去执行其他任务。

Python 的 asyncio 是官方提供的异步 I/O 框架，自 Python 3.4 引入并在 3.5 之后配合 async/await 语法日趋成熟。到了 2026 年，asyncio 已经是 Web 框架（FastAPI、 aiohttp）、数据库驱动、爬虫框架的事实标准。

核心概念速览

理解 asyncio 需要先理清三个核心概念：

协程（Coroutine）：用 async def 定义的函数，调用它不会立即执行，而是返回一个协程对象。

事件循环（Event Loop）：异步任务的调度中心，负责在协程之间切换、管理 I/O 回调。

任务（Task）：被事件循环调度的协程包装对象，可以并发执行并获取结果。

python

import asyncio

async def say_hello(name: str) -> str:
    await asyncio.sleep(0.1)  # 模拟 I/O 等待
    return f"Hello, {name}!"

# 直接调用不会执行，返回协程对象
coro = say_hello("World")
print(coro)  # <coroutine object say_hello at 0x...>

# 用 asyncio.run 启动事件循环并执行
result = asyncio.run(say_hello("Async"))
print(result)  # Hello, Async!

asyncio.run() 是推荐的入口，它会创建一个新的事件循环、运行传入的协程、并在结束后清理资源。一个程序中应尽量避免手动管理多个事件循环。

async/await 语法详解

async def 定义协程函数，await 用于等待一个可等待对象（协程、Task、Future）完成，同时把控制权交还给事件循环。

python

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.2)
    return {"id": user_id, "name": f"user_{user_id}"}

async def fetch_orders(user_id: int) -> list:
    await asyncio.sleep(0.3)
    return [{"order_id": i, "user_id": user_id} for i in range(3)]

async def get_user_detail(user_id: int) -> dict:
    # 串行 await：总耗时约 0.5 秒
    user = await fetch_user(user_id)
    orders = await fetch_orders(user_id)
    return {**user, "orders": orders}

上面是串行写法，两个 await 顺序执行。如果两个请求之间没有依赖关系，就应该并发执行。

并发执行：gather 与 TaskGroup

asyncio.gather 是最常用的并发聚合工具，而 Python 3.11 引入的 asyncio.TaskGroup 提供了更安全、结构化的并发方式。

python

async def get_user_detail_concurrent(user_id: int) -> dict:
    # 并发执行：总耗时约 max(0.2, 0.3) = 0.3 秒
    user, orders = await asyncio.gather(
        fetch_user(user_id),
        fetch_orders(user_id),
    )
    return {**user, "orders": orders}

# Python 3.11+ 推荐的 TaskGroup 写法
async def get_user_detail_taskgroup(user_id: int) -> dict:
    async with asyncio.TaskGroup() as tg:
        t_user = tg.create_task(fetch_user(user_id))
        t_orders = tg.create_task(fetch_orders(user_id))
    # 退出 with 块时所有任务已完成
    return {**t_user.result(), "orders": t_orders.result()}

TaskGroup 的优势在于：任一子任务抛出异常时，它会自动取消其他未完成的任务，避免任务泄漏。在 2026 年的新项目中，优先使用 TaskGroup。

超时与取消控制

网络请求永远不可靠，必须设置超时。Python 3.11 提供了 asyncio.timeout 上下文管理器。

python

async def fetch_with_timeout(url: str) -> str:
    try:
        async with asyncio.timeout(1.0):
            # 假设这是一个慢请求
            await asyncio.sleep(2)
            return f"data from {url}"
    except TimeoutError:
        print(f"请求 {url} 超时")
        return ""

async def main():
    # wait_for 仍然可用，但 timeout 上下文更直观
    result = await asyncio.wait_for(fetch_with_timeout("api"), timeout=1.5)

取消协程时，事件循环会向协程抛入 asyncio.CancelledError。你应该在 finally 块中清理资源。

并发限流：Semaphore

当需要并发请求成百上千个 URL 时，直接 gather 会瞬间打爆目标服务器或本机连接数。用 Semaphore 控制并发上限。

python

import aiohttp

async def fetch_one(session, sem, url):
    async with sem:
        async with session.get(url) as resp:
            return await resp.text()

async def crawl(urls: list[str], concurrency: int = 20) -> list[str]:
    sem = asyncio.Semaphore(concurrency)
    timeout = aiohttp.ClientTimeout(total=10)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        tasks = [fetch_one(session, sem, u) for u in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

return_exceptions=True 让单个失败不会中断整体，异常会作为结果返回，便于后续统计。

实战：异步爬虫完整示例

下面是一个完整的小型异步爬虫，演示了并发、限流、超时、异常处理的组合用法。

python

import asyncio
import aiohttp

async def fetch_title(session, sem, url):
    async with sem:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=8)) as r:
                html = await r.text()
                # 简单提取 <title>
                start = html.find("<title>")
                end = html.find("</title>")
                if start != -1 and end != -1:
                    return url, html[start + 7:end].strip()
                return url, "(no title)"
        except Exception as e:
            return url, f"ERROR: {type(e).__name__}"

async def main(urls):
    sem = asyncio.Semaphore(15)
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(
            *[fetch_title(session, sem, u) for u in urls]
        )
    for url, title in results:
        print(f"{url} -> {title}")

if __name__ == "__main__":
    urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(30)]
    asyncio.run(main(urls))

常见陷阱与最佳实践

不要在异步代码中阻塞事件循环。time.sleep、requests.get、CPU 密集计算都会卡住整个循环。CPU 密集任务用 asyncio.to_thread 或 run_in_executor 丢到线程池。

正确关闭资源。ClientSession、数据库连接池等一定要用 async with 管理。

避免混用同步框架。在 asyncio 中混用同步数据库驱动会失去并发优势，应选择 asyncpg、aiomysql 等异步驱动。

调试未等待的协程。运行时加 -X dev 或设置 PYTHONASYNCIODEBUG=1，可以打印出“协程从未被 await”的警告。

python

async def cpu_heavy(n):
    # 错误：会阻塞事件循环
    # s = sum(i*i for i in range(n))
    # 正确：丢到线程池
    return await asyncio.to_thread(lambda: sum(i * i for i in range(n)))

小结

asyncio 的核心思想是用单线程 + 事件循环实现高并发，特别适合 I/O 密集型任务。掌握 async/await 语法、gather / TaskGroup 并发、Semaphore 限流和超时控制，你就能应对绝大多数异步场景。记住一句话：异步不是多线程，它的力量来自“等待时不闲着”。在实际项目中，配合 FastAPI、aiohttp 等异步生态，可以轻松构建高性能的后端服务。

为什么 Python 需要异步编程

核心概念速览

async/await 语法详解

并发执行：gather 与 TaskGroup

超时与取消控制

并发限流：Semaphore

实战：异步爬虫完整示例

常见陷阱与最佳实践

小结

💬 评论区 (0)