Python 异步编程实战：用 asyncio 让你的网络请求提速 7 倍

前言

在 Web 全栈开发中，网络请求是不可避免的性能瓶颈。假设你需要从 10 个 API 接口拉取数据，每个接口耗时约 1 秒——同步方式需要 10 秒，而使用 Python 的 asyncio 异步编程，只需要 1 秒多。

本文将通过一个完整的实战案例，带你从零掌握 Python 异步编程的核心概念和最佳实践。

---

一、为什么需要异步编程？

1.1 三种并发模型

Python 提供了三种并发编程方式：

模型	适用场景	是否受 GIL 限制
threading（多线程）	I/O 密集型	✅ 是
multiprocessing（多进程）	CPU 密集型	❌ 否
asyncio（异步协程）	网络 I/O 密集型	✅ 是（但无影响）

关键认知：网络请求的大部分时间都在等待响应，CPU 几乎空闲。异步编程就是利用这段等待时间去处理其他请求——而不是傻等。

1.2 同步 vs 异步：一张图看懂

同步（顺序执行）：
  请求1 ████████░░░░ (等待) ████████
  请求2                                 ████████░░░░ (等待) ████████
  请求3                                                                    ████████░░░░...
  总耗时 ≈ 10 秒

异步（并发执行）：
  请求1 ████████░░░░ (等待) ████████
  请求2 ████████░░░░ (等待) ████████
  请求3 ████████░░░░ (等待) ████████
  总耗时 ≈ 1.5 秒 ← 几乎等于最慢那个请求的时间！

---

二、核心概念速览

在写代码之前，先理解几个关键概念：

2.1 async def 和 await

# 普通函数
def normal_func():
    return "hello"

# 协程函数（async def）
async def async_func():
    await asyncio.sleep(1)   # 模拟异步等待
    return "hello"

• async def：定义一个协程函数，调用它不会立即执行，而是返回一个协程对象
• await：暂停当前协程，把控制权交还给事件循环，让其他协程先跑

如果你有 JavaScript 经验：Python 的 async/await 和 JS 的 async/await 概念几乎一样，只是运行环境从浏览器的事件循环换成了 Python 的 asyncio 事件循环。

2.2 事件循环（Event Loop）

事件循环是异步编程的"调度中心"——它维护一个任务队列，哪个协程在等待 I/O，就切换到另一个协程执行。

import asyncio

async def main():
    result = await async_func()

# Python 3.7+ 的标准入口
asyncio.run(main())

---

三、实战：并发抓取 10 个 URL

3.1 模拟场景

我们写一个 fetch_url 函数来模拟网络请求：

import asyncio
import random

async def fetch_url(url: str) -> str:
    """模拟网络请求，延迟 0.5~1.5 秒，10% 概率失败"""
    delay = random.uniform(0.5, 1.5)
    await asyncio.sleep(delay)

    if random.random() < 0.1:
        raise ConnectionError(f"请求超时: {url}")

    return f"<{url}> 响应 (耗时 {delay:.2f}s)"

3.2 ❌ 错误示范：顺序执行

async def sequential_fetch(urls):
    results = []
    for url in urls:
        result = await fetch_url(url)  # 一个接一个等
        results.append(result)
    return results

# 耗时: ~10 秒

这就是"披着 async 外衣的同步代码"——for 循环里逐个 await，本质上还是顺序执行。

3.3 ✅ 正确姿势：asyncio.gather + wrapper 模式

async def async_fetch_all(urls: list[str]) -> tuple[list[str], list[str]]:
    """并发抓取所有 URL，单个失败不影响其他"""

    # Wrapper：在协程内部捕获异常，避免一个失败导致全部崩溃
    async def safe_fetch(url: str):
        try:
            result = await fetch_url(url)
            return ("ok", result, url)
        except ConnectionError:
            return ("fail", None, url)

    # 一次性发起所有请求
    results = await asyncio.gather(
        *[safe_fetch(url) for url in urls]
    )

    # 拆分成功/失败
    success = []
    failed = []
    for status, result, url in results:
        if status == "ok":
            success.append(result)
        else:
            failed.append(url)

    return success, failed

3.4 性能对比

import time

# 同步版本
start = time.perf_counter()
asyncio.run(sequential_fetch(test_urls))
sync_time = time.perf_counter() - start
print(f"同步耗时: {sync_time:.2f}s")

# 异步版本
start = time.perf_counter()
success, failed = asyncio.run(async_fetch_all(test_urls))
async_time = time.perf_counter() - start
print(f"异步耗时: {async_time:.2f}s")
print(f"提速: {sync_time / async_time:.1f}x")

实际运行结果：

同步耗时: 10.52s
异步耗时: 1.37s
提速: 7.7x  ← 10 个请求几乎同时完成！

---

四、关键模式：Wrapper 模式详解

4.1 为什么需要 Wrapper？

asyncio.gather 默认行为：如果其中任何一个协程抛出异常，整个 gather 会立即抛出异常，其他协程的结果全部丢失。

# ❌ 危险写法：一个失败，全盘皆输
results = await asyncio.gather(*[fetch_url(u) for u in urls])
# 如果第 3 个 URL 挂了，前 2 个成功的结果也拿不到！

4.2 Wrapper 模式

async def safe_fetch(url):
    try:
        result = await fetch_url(url)
        return ("ok", result, url)     # 成功：返回成功标记
    except ConnectionError:
        return ("fail", None, url)     # 失败：返回失败标记

results = await asyncio.gather(*[safe_fetch(u) for u in urls])
# 现在每个结果都带有状态标记，不会丢失

核心思想：把异常变成返回值，让业务逻辑自己决定如何处理失败。

4.3 TaskGroup vs gather + wrapper

Python 3.11 引入了 asyncio.TaskGroup，它在上层收集异常并抛出 ExceptionGroup：

方案	优点	缺点
gather + wrapper	API 简洁，返回结果列表	需要手动包装异常
TaskGroup	结构化并发，更现代	失败抛出 ExceptionGroup，处理复杂

建议：初学者从 gather + wrapper 开始，理解后再尝试 TaskGroup。

---

五、完整代码

"""
Python 异步编程实战：并发抓取 URL
运行：python async_fetch.py
"""
import asyncio
import random
import time


async def fetch_url(url: str) -> str:
    """模拟网络请求（0.5~1.5s 延迟，10% 失败率）"""
    delay = random.uniform(0.5, 1.5)
    await asyncio.sleep(delay)
    if random.random() < 0.1:
        raise ConnectionError(f"请求超时: {url}")
    return f"<{url}> 响应 (耗时 {delay:.2f}s)"


async def async_fetch_all(urls: list[str]) -> tuple[list[str], list[str]]:
    """并发抓取所有 URL，单个失败不影响其他"""

    async def safe_fetch(url: str):
        try:
            result = await fetch_url(url)
            return ("ok", result, url)
        except ConnectionError:
            return ("fail", None, url)

    results = await asyncio.gather(
        *[safe_fetch(url) for url in urls]
    )

    success, failed = [], []
    for status, result, url in results:
        if status == "ok":
            success.append(result)
        else:
            failed.append(url)
    return success, failed


if __name__ == "__main__":
    urls = [f"https://api.example.com/item/{i}" for i in range(1, 11)]

    start = time.perf_counter()
    success, failed = asyncio.run(async_fetch_all(urls))
    elapsed = time.perf_counter() - start

    print(f"✅ 成功: {len(success)} 条")
    print(f"❌ 失败: {len(failed)} 条")
    print(f"⏱ 耗时: {elapsed:.2f}s (同步需 ~10s)")

---

六、总结

要点	说明
async/await	Python 协程语法，async def 定义协程，await 挂起等待
事件循环	asyncio.run() 启动，负责调度协程
asyncio.gather	并发执行多个协程，返回结果列表
Wrapper 模式	协程内部 try/except，异常变返回值——单个失败不影响其他
适用场景	网络请求、文件 I/O、数据库查询等 I/O 密集型任务
不适用场景	CPU 密集型计算（用 multiprocessing 代替）

Python 的异步编程上手不难，难的是理解"什么时候该用、怎么用得对"。希望这篇文章能帮你少走弯路。

---

本文基于 Python 3.10+ 语法编写，完整代码可直接运行。