作为一门解释型、动态类型语言，Python 的开发效率毋庸置疑。但每当面对高并发、高 I/O 的场景时，许多人就会抱怨 Python 的速度慢。实际上，这往往是因为没有选对并发模型。

今天，我们将深入探讨 Python 现代并发的基石——Asyncio（异步协程），看看它是如何在单线程下实现高并发的。

一、 传统多线程的软肋：GIL 与上下文切换

在开始协程之前，我们必须先聊聊 Python 开发者绕不开的痛——全局解释器锁（GIL，Global Interpreter Lock）。

1. 什么是 GIL？

由于 CPython 解释器的内存管理不是线程安全的，为了防止多线程并发修改同一个对象，CPython 引入了 GIL。这意味着：在任意时刻，只有一个线程能在 CPU 上运行。

对于计算密集型（CPU-bound）任务，多线程不仅无法利用多核性能，反而会因为频繁的线程上下文切换（Context Switch）导致运行速度变慢。

2. I/O 密集型任务的解法

对于I/O 密集型（I/O-bound）任务（如网络请求、数据库读写），当一个线程在等待网络响应时，它会释放 GIL，让其他线程运行。因此，多线程在 I/O 密集场景是有用的。但线程是非常昂贵的系统资源（每个线程默认占用几兆内存），当并发量达到几千、上万时，操作系统的调度开销就会拖垮整台服务器。

二、 协程的崛起：非阻塞 I/O 与事件循环

协程（Coroutine）是一种用户态的轻量级线程。它的调度完全由用户（代码）控制，而不是由操作系统内核控制。协程的切换几乎没有开销，可以在单线程内轻松创建数十万个协程。

事件循环（Event Loop）工作原理

协程的核心是事件循环。它就像一个永不停歇的传送带：

+-------------------------------------------------+
|                  Event Loop                     |
|                                                 |
|  [Task 1 (Waiting I/O)] ---> [Task 2 (Running)] |
|          ^                         |            |
|          |                         v            |
|     (I/O Finished)           (Yield Control)    |
+-------------------------------------------------+

当某个异步任务遇到 I/O 操作时，它会主动挂起（Yield）并交出 CPU 控制权，事件循环立刻去执行其他就绪的任务。当 I/O 操作完成后，事件循环会在下一轮循环中唤醒该任务继续执行。

三、 Python Asyncio 实战演练

Python 3.5 引入了 async/await 关键字，使异步编程的写法变得像同步代码一样直观。

下面我们通过对比实验，来看看同步代码与异步协程在网络请求场景下的效率差异。

1. 传统的同步阻塞写法

Python

import time
import requests

def fetch_data(url):
    print(f"开始请求: {url}")
    response = requests.get(url)
    print(f"请求完成: {url}")
    return len(response.text)

def main_sync():
    urls = [
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/3"
    ]
    start_time = time.time()
    for url in urls:
        fetch_data(url)
    print(f"同步耗时: {time.time() - start_time:.2f} 秒")

if __name__ == "__main__":
    main_sync()

运行结果：耗时约为 6 秒。因为每一次请求都必须等待上一次请求彻底返回。

2. 高效的 Asyncio 异步写法

使用异步 HTTP 库 httpx 或 aiohttp（不要在异步代码里使用阻塞的 requests 库）：

Python

import asyncio
import time
import httpx

async def fetch_data_async(client, url):
    print(f"开始异步请求: {url}")
    # await 会主动交出控制权，让事件循环执行其他请求
    response = await client.get(url)
    print(f"异步请求完成: {url}")
    return len(response.text)

async def main_async():
    urls = [
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/3"
    ]
    start_time = time.time()
    
    # 推荐使用 async with 管理异步客户端生命周期
    async with httpx.AsyncClient() as client:
        # 并发打包所有任务
        tasks = [fetch_data_async(client, url) for url in urls]
        # 并发执行并等待所有任务完成
        results = await asyncio.gather(*tasks)
        
    print(f"异步并发耗时: {time.time() - start_time:.2f} 秒")

if __name__ == "__main__":
    # 启动事件循环
    asyncio.run(main_async())

运行结果：耗时仅约 3 秒。三大请求是几乎同时发出的，时间取决于最慢的那个请求（3秒）。

四、 避坑指南：写异步代码时的常见致命错误

1. 混用阻塞库：在异步函数中使用 time.sleep()、requests.get() 或进行繁重的 CPU 密集计算。这会直接卡死整个事件循环。正确的做法是使用 await asyncio.sleep()，或者将计算密集任务通过 run_in_executor 扔进进程池。

2. 忘记使用 await：调用一个 async 函数不会直接执行它，而是会返回一个协程对象（Coroutine Object）。你必须使用 await 或者将其放入事件循环中才能执行。