asyncio事件循环源码级分析与实战

引言

去年双十一大促,我负责的一个订单查询服务突然出现了诡异的性能问题:QPS 从平时的 8000 骤降到 1200,但 CPU 利用率却只有 30%,内存也正常。排查了整整一天,最后定位到一段看起来人畜无害的代码:

async def get_order_detail(order_id: int):
    # 缓存没命中,走 DB
    order = await db.fetch_one("SELECT * FROM orders WHERE id = %s", order_id)
    # 这里调用了同步的 requests 库!
    user_info = requests.get(f"http://user-service/users/{order.user_id}").json()
    return {**order, "user": user_info}

问题就出在这一行 requests.get。它是个同步阻塞调用,直接卡死了整个事件循环。一个请求卡 50ms,后面所有协程都得排队等待。这就是典型的"用 asyncio 写同步代码"——你以为你在开跑车,其实你在推独轮车。

这个坑之所以常见,是因为大多数人对 asyncio 的理解停留在 async/await 语法层面,而不知道事件循环底下到底发生了什么。这篇文章我会带你从源码级别拆解 asyncio 事件循环,搞清楚 await 背后到底发生了什么、为什么一个同步调用能毁掉整个服务、以及如何写出真正高性能的异步代码。

---

核心概念:事件循环到底是什么

生活类比:餐厅的"万能服务员"

想象一个只有一个服务员的餐厅:

• 同步阻塞模式(线程池):客人 A 点完菜,服务员必须站在厨房门口等菜做好、端上桌,才能去服务客人 B。为了同时服务多人,餐厅只能雇很多服务员(多线程),但每个服务员大部分时间都在"干等"。

• 异步事件循环模式:服务员点完 A 的菜,把单子扔给厨房(注册回调),立刻去服务 B;谁的菜好了厨房喊一声(事件就绪),服务员才去端。一个服务员就能服务几十桌客人,因为他从不在"等待"上浪费时间。

这个"服务员"就是事件循环(Event Loop),"厨房喊一声"就是 I/O 就绪事件,"把单子扔给厨房"就是注册回调/挂起协程。

技术定义

事件循环本质上是一个单线程的无限循环,它不断做三件事:

1. 从就绪队列取出已完成的 I/O 事件对应的回调/协程,执行它们;
2. 检查有没有定时器到期,执行对应的回调;
3. 把所有未就绪的 I/O 注册到操作系统的多路复用器(epoll/kqueue/IOCP)上,阻塞等待,直到有新事件就绪。

关键点在于第 3 步:阻塞是发生在"整个循环"上的,而不是某个协程上。当循环阻塞在 epoll_wait 时,只要有一个 socket 可读,它就立刻返回,唤醒对应协程继续执行。这就是"单线程处理海量并发"的秘密。

三个核心角色的关系

graph TD A[EventLoop 事件循环] -->|调度| B[Task 任务] B -->|驱动| C[Coroutine 协程] C -->|yield 出 Future| B B -->|注册回调| A A -->|epoll_wait| D[Selector 多路复用器] D -->|I/O 就绪| A A -->|set_result| E[Future 未来对象] E -->|唤醒| C

• Coroutine(协程):用 async def 定义的函数,是一个可以被暂停/恢复的执行体。

• Future(未来对象):一个"占位符",代表"未来会有个结果"。协程 await 一个 Future 时,就是把自己挂在它上面,等结果来了再被唤醒。

• Task(任务):Future 的子类,负责驱动协程——它把协程包起来扔进事件循环,协程 yield 出的 Future 由 Task 接管。

记住这个链条:事件循环驱动 Task,Task 驱动 Coroutine,Coroutine yield 出 Future,事件循环监听 Future 对应的 I/O,就绪后 set_result 唤醒协程。理解了这条链,后面的源码就都好懂了。

---

源码级深度分析

从 asyncio.run 开始

一切的入口是 asyncio.run(main())。我们看 CPython 3.11 的源码(位于 Lib/asyncio/runners.py,部分为 C 实现,这里用 Python 版逻辑):

def run(main, *, debug=None):
    if events._get_running_loop() is not None:
        raise RuntimeError("asyncio.run() cannot be called from a running event loop")
    # 1. 创建事件循环
    loop = events.new_event_loop()
    try:
        events.set_event_loop(loop)
        # 2. 把 main 协程包成 Task,跑起来直到完成
        return loop.run_until_complete(main)
    finally:
        try:
            _cancel_all_tasks(loop)      # 3. 取消所有残留任务
            loop.run_until_complete(loop.shutdown_asyncgens())
            loop.run_until_complete(loop.shutdown_default_executor())
        finally:
            events.set_event_loop(None)
            loop.close()

核心在 loop.run_until_complete(main)。它的作用是:把协程包成 Task,然后驱动事件循环一直跑,直到这个 Task 完成。

run_until_complete 与 run_forever

# Lib/asyncio/base_events.py
def run_until_complete(self, future):
    self._check_closed()
    self._check_running()
    new_task = not futures.isfuture(future)
    # 关键:ensure_future 把协程包装成 Task
    future = tasks.ensure_future(future, loop=self)
    if new_task:
        future._log_destroy_pending = False
    # 添加完成回调:Task 完成时停止循环
    future.add_done_callback(_run_until_complete_cb)
    try:
        self.run_forever()
    except:
        if new_task and future.done() and not future.cancelled():
            future.exception()
        raise
    finally:
        future.remove_done_callback(_run_until_complete_cb)
    if not future.done():
        raise RuntimeError('Event loop stopped before Future completed.')
    return future.result()

_run_until_complete_cb 就是那个"停止开关":当 Task 完成时调用 loop.stop(),让 run_forever 退出。

事件循环的心脏:_run_once

run_forever 内部是一个无限循环,每次循环调用 _run_once()。这是整个 asyncio 的心脏,每一行都值得细看:

# Lib/asyncio/base_events.py (精简注释版)
def _run_once(self):
    # ---- 阶段 1:计算本次 select 的超时时间 ----
    sched_count = len(self._scheduled)   # 定时器堆(最小堆)
    if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
            self._timer_cancelled_count / sched_count > _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
        # 取消的定时器太多,堆里垃圾比例过高,重建堆
        new_scheduled = []
        for handle in self._scheduled:
            if handle._cancelled:
                handle._scheduled = False
            else:
                new_scheduled.append(handle)
        heapq.heapify(new_scheduled)
        self._scheduled = new_scheduled
        self._timer_cancelled_count = 0
    else:
        # 惰性清理:从堆顶弹出已取消的定时器
        while self._scheduled and self._scheduled[0]._cancelled:
            self._timer_cancelled_count -= 1
            handle = heapq.heappop(self._scheduled)
            handle._scheduled = False

    timeout = None
    if self._ready or self._stopping:
        timeout = 0                          # 有就绪回调,别阻塞,立即处理
    elif self._scheduled:
        # 用最近的定时器时间减去当前时间,作为 select 的超时
        when = self._scheduled[0]._when
        timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT)

    # ---- 阶段 2:注册 I/O 并阻塞等待事件 ----
    # 底层调用 selector.select(timeout),即 epoll_wait
    event_list = self._selector.select(timeout)
    # 处理 select 返回的就绪事件(会调用注册的回调,进而 set_result 唤醒协程)
    self._process_events(event_list)
    # 处理"立即执行"的回调(如 call_soon 注册的)
    ntodo = len(self._ready)
    for i in range(ntodo):
        handle = self._ready.popleft()
        if handle._cancelled:
            continue
        if self._debug:
            # ... debug 相关
            handle._run()
        else:
            handle._run()                    # 执行回调,恢复协程

    # ---- 阶段 3:执行到期定时器 ----
    end_time = self.time() + self._clock_resolution
    while self._scheduled:
        handle = self._scheduled[0]
        if handle._when >= end_time:
            break
        handle = heapq.heappop(self._scheduled)
        handle._scheduled = False
        self._ready.append(handle)           # 到期的定时器丢进就绪队列

这里有三个关键点:

1. self._ready 和 self._scheduled 是两个不同的队列。 _ready 是"马上就要执行的回调"(双端队列),_scheduled 是"未来某个时刻要执行的回调"(最小堆实现的定时器)。

2. select(timeout) 是唯一的阻塞点。 它的 timeout 是精心计算的:有就绪回调就传 0(不阻塞),否则用最近定时器的时间。这样既保证不饿死 I/O,又能及时触发定时器。

3. 关于 _ready 的"两轮循环"陷阱。 注意 ntodo = len(self._ready),它只处理进入循环时快照的长度。这意味着:如果一个回调又往 _ready 里塞了新回调,新回调要等到下一轮 _run_once 才执行。这个设计是为了防止回调无限自我递归导致循环卡死。

await 背后发生了什么

现在看最关键的问题:await 到底干了啥?以 await asyncio.sleep(1) 为例。

Task 的 __step 方法是驱动协程的核心(Task 继承自 C 实现的 _asyncio.Task,逻辑在 C 里,但等价于下面这段 Python 逻辑):

# Task.__step 的简化逻辑(实际在 C 中实现)
def __step(self, exc=None):
    coro = self._coro
    try:
        if exc is None:
            # 驱动协程向前跑一步,拿到它 yield 出来的东西
            result = coro.send(None)
        else:
            result = coro.throw(exc)
    except StopIteration as exc:
        # 协程正常结束
        super().set_result(exc.value)
    else:
        # 协程 yield 出了一个 Future(await 的对象)
        if isinstance(result, Future):
            result.add_done_callback(self.__wakeup)   # 挂上唤醒回调!
        else:
            self._loop.call_soon(self.__step)          # 否则下一轮继续跑

def __wakeup(self, future):
    # Future 完成时被调用,带着结果回到 __step
    try:
        future.result()   # 若 Future 有异常,这里抛出
    except Exception as exc:
        self.__step(exc)
    else:
        self.__step()

再结合 asyncio.sleep 的实现:

# Lib/asyncio/tasks.py
async def sleep(delay, result=None):
    if delay <= 0:
        await __sleep0()
        return result
    loop = events.get_running_loop()
    future = loop.create_future()               # 创建一个 Future
    h = loop.call_later(delay, futures._set_result_unless_cancelled, future, result)
    try:
        return await future                     # 挂起,等 Future 有结果
    finally:
        h.cancel()

现在把整条链路串起来:

1. 协程执行到 await future,Python 解释器把这个 Future yield 给 Task(通过 coro.send(None) 返回);
2. Task 收到 Future,给它挂上 __wakeup 回调,然后返回,控制权交回事件循环;
3. 事件循环继续 _run_once,把 call_later 注册的定时器放进 _scheduled;
4. 1 秒后定时器到期,事件循环执行 _set_result_unless_cancelled,即 future.set_result(result);
5. set_result 触发 Future 上挂的所有回调——包括 Task 的 __wakeup;
6. __wakeup 调用 __step,coro.send(None) 让协程从 await 处恢复执行。

"await 一个 Future"的本质,就是"把自己挂到 Future 的回调链上,然后让出 CPU 给事件循环"。 这也解释了为什么 await 只能是异步的——它需要把控制权交还给事件循环,而不是像普通函数调用一样立即返回。

为什么同步阻塞会毁掉一切

现在回到开头的 bug。当 requests.get() 执行时:

• 它调用的是阻塞式 socket,代码在当前线程里一直卡着等待网络返回;

• 这个线程就是事件循环线程;

• 事件循环线程卡住 → _run_once 无法继续 → epoll_wait 无法被调用 → 所有其他协程全部冻住。

类比:整个餐厅只有一个服务员,他跑到厨房门口站着等一道菜,其他几十桌客人全都被晾着。这就是为什么一个 50ms 的同步调用能让 QPS 断崖式下跌。

---

实战代码

示例 1:手写一个迷你事件循环

要真正理解 asyncio,最好的办法是自己写一个。下面这个 80 行的迷你事件循环,用 selectors 实现,能跑真正的异步 I/O。注意:此示例兼容 asyncio 的协程语法,但使用自定义调度器,不能与 asyncio 混用。

import selectors
import socket
import time
import heapq
from collections import deque

class MiniEventLoop:
    """一个极简事件循环,演示 asyncio 的核心调度逻辑"""

    def __init__(self):
        self._selector = selectors.DefaultSelector()
        self._ready = deque()          # 就绪回调队列
        self._scheduled = []           # 定时器最小堆
        self._stopping = False

    def call_soon(self, callback, *args):
        """注册一个立即执行的回调"""
        self._ready.append((callback, args))

    def call_later(self, delay, callback, *args):
        """注册一个延迟执行的回调,返回可取消的句柄"""
        when = time.monotonic() + delay
        handle = [when, callback, args, False]
        heapq.heappush(self._scheduled, handle)
        return handle

    def _run_once(self):
        """核心:一次循环迭代"""
        # 1. 计算 select 超时
        timeout = 0 if self._ready else None
        if self._scheduled:
            # 清理已取消的定时器
            while self._scheduled and self._scheduled[0][3]:
                heapq.heappop(self._scheduled)
            if self._scheduled:
                when = self._scheduled[0][0]
                timeout = max(0, when - time.monotonic())

        # 2. 阻塞等待 I/O 事件(或超时)
        events = self._selector.select(timeout)
        for key, mask in events:
            callback, args = key.data
            self._ready.append((callback, args))

        # 3. 执行就绪回调(注意只处理本轮快照,防止无限递归)
        ntodo = len(self._ready)
        for _ in range(ntodo):
            callback, args = self._ready.popleft()
            callback(*args)

        # 4. 执行到期定时器
        now = time.monotonic()
        while self._scheduled and self._scheduled[0][0] <= now:
            handle = heapq.heappop(self._scheduled)
            if not handle[3]:
                self._ready.append((handle[1], handle[2]))

    def add_reader(self, fd, callback, *args):
        """注册读事件"""
        self._selector.register(fd, selectors.EVENT_READ, (callback, args))

    def run_forever(self):
        while not self._stopping:
            self._run_once()

    def stop(self):
        self._stopping = True


# ---- 用迷你循环实现一个异步 HTTP GET ----
def async_fetch(loop, host, path, port=80):
    """返回一个生成器,配合 loop 调度实现异步 HTTP 请求"""
    sock = socket.socket()
    sock.setblocking(False)
    try:
        sock.connect((host, port))
    except BlockingIOError:
        pass  # 非阻塞 connect 会立即抛异常,这是正常的

    # 等待可写(连接建立完成)
    fut = {'done': False}
    def on_writable():
        loop._selector.unregister(sock)
        sock.sendall(f"GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n".encode())
        # 连接建立后,注册读事件
        def on_readable():
            data = b""
            while True:
                try:
                    chunk = sock.recv(4096)
                except BlockingIOError:
                    break
                if not chunk:
                    break
                data += chunk
            print(f"[{host}] 收到 {len(data)} 字节")
            fut['done'] = True
            loop._selector.unregister(sock)
            sock.close()
        loop.add_reader(sock.fileno(), on_readable)

    loop._selector.register(sock, selectors.EVENT_WRITE, (on_writable, ()))
    return fut


if __name__ == "__main__":
    loop = MiniEventLoop()
    # 同时发起多个请求,验证并发(注意:示例用明文 HTTP,仅用于演示)
    targets = [
        ("example.com", "/"),
        ("www.python.org", "/"),
    ]
    pending = [async_fetch(loop, h, p) for h, p in targets]

    def check_done():
        if all(f['done'] for f in pending):
            loop.stop()
        else:
            loop.call_later(0.05, check_done)

    loop.call_later(0.05, check_done)
    loop.run_forever()
    print("全部完成")

这个迷你循环虽然简陋,但和 asyncio 的核心结构一一对应:_ready/_scheduled 双队列、select 阻塞、快照式回调处理。跑一遍你会对事件循环有完全不同的体感。

示例 2:用 loop.run_in_executor 安全地包装同步调用

回到开头的 bug。正确的做法是把同步阻塞调用扔到线程池里,让事件循环不被卡住:

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def blocking_io(task_id: int) -> str:
    """模拟一个同步阻塞操作,比如 requests.get 或慢查询"""
    time.sleep(1)  # 阻塞 1 秒
    return f"task-{task_id} done at {time.time():.2f}"

async def wrong_way():
    """错误示范:直接在协程里调用阻塞函数,事件循环被卡死"""
    start = time.time()
    results = [blocking_io(i) for i in range(5)]  # 串行阻塞 5 秒
    print(f"[错误] 耗时 {time.time() - start:.2f}s,结果: {results}")

async def right_way():
    """正确示范:用 run_in_executor 扔到线程池,并发执行"""
    loop = asyncio.get_running_loop()
    start = time.time()
    # 用线程池执行同步函数,事件循环可以继续调度其他协程
    with ThreadPoolExecutor(max_workers=5) as pool:
        tasks = [
            loop.run_in_executor(pool, blocking_io, i)
            for i in range(5)
        ]
        results = await asyncio.gather(*tasks)
    print(f"[正确] 耗时 {time.time() - start:.2f}s,结果: {results}")

async def main():
    print("=== 错误示范 ===")
    await wrong_way()
    print("=== 正确示范 ===")
    await right_way()

if __name__ == "__main__":
    asyncio.run(main())

输出:

=== 错误示范 ===
[错误] 耗时 5.00s,结果: [...]
=== 正确示范 ===
[正确] 耗时 1.01s,结果: [...]

run_in_executor 的本质是把同步函数提交给线程池,返回一个 concurrent.futures.Future,asyncio 再把它包装成 asyncio.Future 挂到事件循环上。这样阻塞发生在工作线程里,事件循环线程依然能自由调度。

注意坑点:run_in_executor 默认使用的线程池大小是 min(32, os.cpu_count() + 4),如果并发量很大而每个同步调用都很慢,线程池会成为新的瓶颈。这种场景要考虑换成真正的异步库(如 httpx、asyncpg)。

示例 3:自定义协议实现一个异步 TCP echo 服务

最后来个硬核的:用 asyncio.Protocol(基于回调的底层 API)实现一个 TCP echo 服务,对比 StreamReader/StreamWriter 的高层 API。

import asyncio

class EchoProtocol(asyncio.Protocol):
    """自定义协议:基于回调,零协程开销,适合高吞吐场景"""

    def __init__(self):
        self.transport = None
        self.peername = None
        self.buffer = b""

    def connection_made(self, transport):
        """连接建立时回调"""
        self.transport = transport
        self.peername = transport.get_extra_info('peername')
        print(f"[+] 新连接: {self.peername}")

    def data_received(self, data):
        """收到数据时回调 —— 这里是性能关键路径"""
        self.buffer += data
        # 简单的行协议:按换行符切分
        while b"\n" in self.buffer:
            line, self.buffer = self.buffer.split(b"\n", 1)
            # 回显并加上前缀,注意:这里没有 await,没有协程切换开销
            self.transport.write(b"echo: " + line + b"\n")

    def connection_lost(self, exc):
        """连接断开时回调"""
        print(f"[-] 连接关闭: {self.peername}, exc={exc}")

    def eof_received(self):
        self.transport.close()
        return False


async def main():
    loop = asyncio.get_running_loop()
    # create_server 内部会为每个连接创建一个 Protocol 实例
    server = await loop.create_server(
        EchoProtocol, '127.0.0.1', 8888
    )
    print("Echo 服务启动: 127.0.0.1:8888")
    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("服务停止")

用 nc localhost 8888 连上去输入几行文字,你会看到即时回显。

为什么用 Protocol 而不是 Stream API? StreamReader/StreamWriter 是基于 Protocol 封装的高层 API,每次 read/write 都涉及协程调度,适合业务代码可读性。而 Protocol 是纯回调,没有协程切换开销,在需要处理海量短连接(如网关、代理)时性能更高。这就是"高层 API 换可读性、底层 API 换性能"的经典权衡。

---

方案对比:asyncio vs 其他并发模型

维度	多线程	多进程	asyncio	协程库(gevent)
并发单元	线程	进程	协程	协程(monkey patch)
上下文切换开销	中(内核态)	高(进程切换)	极低(用户态)	极低
内存占用	每线程 ~8MB 栈	每进程独立内存	每协程 ~几 KB	每协程 ~几 KB
GIL 影响	受限(CPU 密集无效)	无	受限(但 I/O 密集无妨)	受限
编程模型	同步、直观	同步	async/await、传染性	同步代码、隐式切换
调试难度	低	中	高(堆栈断裂)	中
生态兼容	全兼容	全兼容	需异步库	需 monkey patch
适用场景	I/O 密集、CPU 中等	CPU 密集	高并发 I/O	存量同步代码改造

关键结论:

• CPU 密集型:老老实实上多进程(multiprocessing)或 ProcessPoolExecutor,asyncio 帮不了你,因为 GIL 还在。

• I/O 密集型、超大规模并发:asyncio 是最优解,一个进程扛几万连接是常态。FastAPI、aiohttp 这类框架就是靠它撑起高 QPS。

• 存量同步代码:如果不想大改,gevent 的 monkey patch 能让同步代码"自动"变异步,但代价是隐式切换、调试困难,而且和 C 扩展库兼容性差。新项目不建议。

• Django 场景:Django 3.1+ 支持 async view,但 ORM 仍是同步的(4.x 才逐步提供异步接口)。在 Django 里写 async view 时,务必用 sync_to_async 包装 ORM 调用,否则就是开头那个 bug 的翻版。FastAPI 则原生异步,搭配 asyncpg/SQLAlchemy 2.0 异步引擎更丝滑。

---

最佳实践与避坑指南

1. 永远不要在协程里直接调用同步阻塞函数

这是头号杀手。常见黑名单:requests、time.sleep、open() 大文件读写、同步 DB 驱动(pymysql、psycopg2)、subprocess.run。

替代方案:

• HTTP → httpx / aiohttp

• DB → asyncpg / aiomysql / SQLAlchemy 2.0 async

• 睡眠 → asyncio.sleep

• 文件 → aiofiles

• 实在没有异步库 → loop.run_in_executor

2. 不要创建"僵尸任务"

# 反例:创建了任务却不等待,异常被吞掉
async def bad():
    asyncio.create_task(risky_operation())  # 异常无人处理

# 正例:保存引用并在合适的时机 await
async def good():
    task = asyncio.create_task(risky_operation())
    # ... 做别的事
    await task   # 或 task.add_done_callback(handle_exception)

CPython 官方文档明确警告:事件循环只持有任务的弱引用,如果任务对象被 GC 回收,任务会被静默取消。所以务必保存强引用(比如放进一个 set)。

3. gather 的异常传播陷阱

# 默认行为:一个失败,其他会被取消(但不会立即停止)
results = await asyncio.gather(*tasks)

# 推荐:用 return_exceptions=True,自己处理每个异常
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        logger.error("任务失败", exc_info=r)

4. 优雅关闭:处理 CancelledError

async def worker():
    try:
        while True:
            await do_work()
    except asyncio.CancelledError:
        # 清理资源(关闭连接、回滚事务)
        await cleanup()
        raise  # 必须重新抛出,否则取消信号被吞掉

注意:Python 3.8+ 中 CancelledError 继承自 BaseException 而不是 Exception,所以 except Exception 抓不到它,这是故意设计——防止误吞取消信号。

5. debug 模式揪出慢回调

# 开发环境开启 debug,会打印执行超过 100ms 的回调
asyncio.run(main(), debug=True)

它会输出类似 Executing took 0.152 seconds 的警告,直接帮你定位同步阻塞点。生产环境别开,性能损耗明显。

6. 不要在协程里用 threading.local

协程共享线程,threading.local 在协程间会互相污染。要用 contextvars.ContextVar,它才是协程安全的"上下文变量"。FastAPI 的依赖注入、日志 trace_id 传递都靠它。

---

总结

回到开头那个 bug——一位同事把 requests.get 直接写进了协程,让整个事件循环卡死。这个坑之所以普遍,是因为大家对 asyncio 的认知停留在语法糖层面。这篇文章我们从源码角度把整条链路走了一遍:

1. 事件循环的心脏是 _run_once:它用 _ready 队列处理就绪回调、用 _scheduled 堆管理定时器、用 selector.select 阻塞等待 I/O,三者循环往复。
2. await 的本质是"挂起到 Future 的回调链上":协程 yield 出 Future,Task 给它挂 __wakeup,Future 完成时事件循环触发回调,协程被唤醒。
3. 单线程调度的前提是"永不阻塞":任何同步阻塞调用都会让整个循环停摆,这是 asyncio 性能模型的根本约束,也是最大的坑。

延伸思考:

• uvloop 为什么快? 它用 Cython 重写了事件循环,底层用 libuv(Node.js 同款),epoll 调用和回调调度都比纯 Python 快 2-4 倍。生产环境可以 uvloop.install() 一行提速。

• asyncio 的"结构化并发"演进:Python 3.11 引入了 asyncio.TaskGroup,3.13 引入了 asyncio.timeout,这些是向 Trio 的 nursery 模型靠拢——让任务生命周期有明确的父子关系,避免"僵尸任务"和取消传播混乱。新项目建议优先用 TaskGroup。

• 和 Go 的 goroutine 对比:Go 的调度器是 M:N 模型(多个 goroutine 映射到多个 OS 线程),能自动利用多核;Python asyncio 是 1:N(单线程多协程),受 GIL 限制。这是设计哲学的根本差异——Go 追求"并行 + 并发",asyncio 只追求"并发"。

理解了事件循环,你就理解了 Python 异步编程的一切。下次再看到 await,你脑子里应该浮现的不是一行语法,而是 _run_once 里那个永不疲倦的循环,和它背后 epoll_wait 静静等待的样子。