关注+星标，每天学习Python新技能

因公众号更改推送规则，请点“在看”并加“星标”第一时间获取精彩技术分享

来源于网络，侵删

问题背景

Python 的异步编程经历了三代演进：回调地狱 → yield from → async/await。很多开发者会用 async def 但不理解事件循环（Event Loop）怎么调度协程、await 和 yield 底层是什么关系、为什么 CPU 密集型任务不适合 asyncio。这篇文章从底层原理讲起，用可运行的代码逐层拆解。

▲ Python异步编程三代演进图——回调→yield from→async/await，标注每代的核心改进 

第一层：生成器到协程 — yield 的本质

Python 协程的"基因"来自生成器。理解 yield 是理解 await 的前提。

# 生成器：可以暂停和恢复的函数def simple_gen():    print("Step 1")    yield 1    print("Step 2")    yield 2    print("Done")g = simple_gen()print(next(g))  # 输出: Step 1 → 1  (跑到第一个 yield，暂停)print(next(g))  # 输出: Step 2 → 2  (从上次暂停处继续，跑到下个 yield)# next(g) → StopIteration

生成器有三个关键能力：暂停执行（yield）、恢复执行（next/send）、传值（send 方法）。这三者恰好是协程需要的全部底层能力。

# 用生成器模拟协程调度def coroutine_a():    for i in range(3):        print(f"A: {i}")        yield  # 主动让出执行权def coroutine_b():    for i in range(3):        print(f"B: {i}")        yielddef scheduler(*coros):    """简易协程调度器"""    iters = [c() for c in coros]    while iters:        for it in iters[:]:            try:                next(it)            except StopIteration:                iters.remove(it)scheduler(coroutine_a, coroutine_b)# 输出: A:0 B:0 A:1 B:1 A:2 B:2  (交错执行！)

第二层：事件循环 — asyncio 的心脏

事件循环（Event Loop）本质上是一个"等待事情发生"的循环。它维护一个任务队列，不断检查哪些任务可以继续执行。

import asyncioasync def fetch_data(url, delay):    """模拟网络请求"""    print(f"开始请求 {url}...")    await asyncio.sleep(delay)  # 模拟 I/O 等待    print(f"{url} 响应完成 ({delay}s)")    return f"data from {url}"async def main():    # 并发发起两个请求    start = time.time()    task1 = asyncio.create_task(fetch_data("https://api1.com", 2))    task2 = asyncio.create_task(fetch_data("https://api2.com", 1))    # await 等待结果（这里 task2 会先完成）    result1 = await task1    result2 = await task2    print(f"总耗时: {time.time() - start:.1f}s")  # ~2.0s，不是 3.0s！    print(result1, result2)asyncio.run(main())

关键理解：await asyncio.sleep(2) 不会阻塞线程！它把控制权交还给事件循环，事件循环去执行其他任务。2 秒后事件循环唤醒它继续执行。

事件循环的伪代码实现

class SimpleEventLoop:    def __init__(self):        self.ready = []    # 待执行的任务    def call_soon(self, coro):        """添加一个就绪的协程"""        self.ready.append(coro)    def run_forever(self):        """简化的事件循环"""        while self.ready:            coro = self.ready.pop(0)            try:                # 驱动协程到下一个 await                coro.send(None)                # 如果没抛 StopIteration，说明协程还在等                self.ready.append(coro)            except StopIteration:                pass  # 协程执行完毕

▲ 事件循环调度图——Task1和Task2在EventLoop中交替执行，await时将控制权交还给Loop

第三层：async/await vs yield — 底层关系

# await 本质上是 yield 的语法糖# 下面两段代码等价：# 方式1：await 语法async def fetch():    data = await http_get("/api")    return data# 方式2：yield from 等价（Python 3.4 时代）@asyncio.coroutinedef fetch():    data = yield from http_get("/api")    return data

await 做的事情：① 将当前协程暂停 ② 将控制权交还给事件循环 ③ 等待 awaitable 对象完成后恢复到暂停点继续执行。这和之前 yield 做的"暂停-恢复-传值"完全一致。

第四层：Task — 让协程并发运行

await coro() 是串行等待，create_task() 才是真正的并发：

# 错误写法：串行执行（总耗时 = sum）async def sequential():    r1 = await fetch("url1", 2)  # 等 2s    r2 = await fetch("url2", 1)  # 再等 1s    # 总耗时 3s# 正确写法：并发执行（总耗时 = max）async def concurrent():    task1 = asyncio.create_task(fetch("url1", 2))    task2 = asyncio.create_task(fetch("url2", 1))    r1 = await task1  # 等待时 task2 也在跑    r2 = await task2    # 总耗时 2s（最慢的那个）

Task 的本质：把协程包装成一个 Future 对象，注册到事件循环中，让事件循环管理它的生命周期。

# Task 的简化实现class Task:    def __init__(self, coro):        self.coro = coro        # 被封装的协程        self._done = False        self._result = None    def __await__(self):        """让 await task 可以工作"""        while not self._done:            yield  # 自我暂停，等待被唤醒        return self._result    def _wakeup(self, result):        self._result = result        self._done = True

▲ Task并发执行时序图——Task1和Task2同时注册到EventLoop，各自独立await，最短时间完成最长任务

第五层：实战 — 异步 HTTP 并发抓取

import asyncioimport aiohttpfrom typing import Listasync def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:    """抓取单个 URL"""    try:        async with session.get(url, timeout=10) as resp:            data = await resp.json()            return {"url": url, "status": resp.status, "data": data}    except Exception as e:        return {"url": url, "error": str(e)}async def batch_fetch(urls: List[str], concurrency: int = 10) -> List[dict]:    """并发抓取多个 URL，限制并发数"""    semaphore = asyncio.Semaphore(concurrency)    async def fetch_with_limit(url):        async with semaphore:            async with aiohttp.ClientSession() as session:                return await fetch_url(session, url)    return await asyncio.gather(*[fetch_with_limit(u) for u in urls])# 使用urls = [f"https://jsonplaceholder.typicode.com/todos/{i}" for i in range(50)]results = asyncio.run(batch_fetch(urls, concurrency=20))print(f"完成 {len(results)} 个请求")

面试要点总结

总结

Python 异步编程的核心不是 async/await 语法，而是事件循环 + 协程调度这个底层模型。理解了三件事就掌握了全局：① yield 让函数可以暂停和恢复 ② 事件循环是一个不断检查'谁可以继续执行'的循环 ③ await 就是把控制权交还给事件循环去处理其他任务。记住一个原则：I/O 密集型用 asyncio，CPU 密集型用 multiprocessing，两者混用用 ProcessPoolExecutor + run_in_executor。