30 分钟

Python 高级

asyncio 事件循环深度

理解事件循环原理、Task 调度和异步陷阱

理解事件循环的工作原理
掌握 Task/Future 的状态机
理解 async for/async with 协议
识别并避免异步陷阱

asyncio 事件循环深度

asyncio 是 Python 异步编程的核心。理解事件循环原理，才能写出正确的高性能异步代码。

事件循环原理

事件循环是个无限循环，不停从就绪队列里取回调执行。核心就是等I/O的时候，切去别的任务跑。

示例代码（可运行）

Python 0%

1# 事件循环简化模型

2import asyncio

4async def say_after(delay, msg):

5 await asyncio.sleep(delay)

6 print(msg)

8async def main():

9 # 串行：总时间 2 秒

10 print("=== 串行 ===")

11 start = asyncio.get_event_loop().time()

12 await say_after(1, "hello")

13 await say_after(1, "world")

14 print(f"耗时: {asyncio.get_event_loop().time()-start:.1f}s")

16 # 并发：总时间 1 秒

17 print("\n=== 并发(gather) ===")

18 start = asyncio.get_event_loop().time()

19 await asyncio.gather(

20 say_after(1, "hello"),

21 say_after(1, "world")

22 )

23 print(f"耗时: {asyncio.get_event_loop().time()-start:.1f}s")

25asyncio.run(main())

Task 与 Future

示例代码（可运行）

Python 0%

1# Task 状态机

2import asyncio

4async def worker(name, seconds):

5 print(f"{name} 开始")

6 await asyncio.sleep(seconds)

7 print(f"{name} 完成")

8 return f"{name}的结果"

10async def main():

11 # create_task 立即调度，不等待

12 task1 = asyncio.create_task(worker("A", 1))

13 task2 = asyncio.create_task(worker("B", 1))

15 print("任务已创建，事件循环开始执行")

17 # await 等待结果

18 results = await asyncio.gather(task1, task2)

19 print(f"结果: {results}")

21 # Task 状态

22 task3 = asyncio.create_task(worker("C", 0.5))

23 print(f"done? {task3.done()}") # False

24 await task3

25 print(f"done? {task3.done()}") # True

26 print(f"result: {task3.result()}")

28asyncio.run(main())

异步陷阱：阻塞调用

示例代码（可运行）

Python 0%

1# 陷阱1：在异步代码中调用同步阻塞函数

2import asyncio

3import time

5async def bad_example():

6 # time.sleep 是阻塞的！它会冻结整个事件循环

7 # 所有其他协程都无法执行

8 time.sleep(1) # 错误！

9 await asyncio.sleep(1) # 正确：非阻塞等待

11async def main():

12 # 陷阱2：忘记 await

13 async def coro():

14 return 42

16 # coro() 只是创建协程对象，不会执行！

17 result = coro() # RuntimeWarning: coroutine was never awaited

18 print(result) # <coroutine object>

20 result = await coro() # 正确

21 print(result) # 42

23 # 陷阱3：未保存 Task 引用导致被 GC

24 asyncio.create_task(coro()) # 可能被垃圾回收！

25 task = asyncio.create_task(coro()) # 保存引用

26 await task

28asyncio.run(main())

🚫异步代码最大的坑

在 async 函数中调用任何同步阻塞操作（time.sleep、requests.get、open() 大文件、CPU 密集计算）都会冻结整个事件循环，所有并发都失效。解决方案：用异步替代（aiohttp 替代 requests、asyncio.sleep 替代 time.sleep）用 loop.run_in_executor 把阻塞调用丢到线程池CPU 密集用 ProcessPoolExecutor。

async for / async with

示例代码（可运行）

Python 0%

1# 异步迭代器协议

2import asyncio

4class AsyncCounter:

5 def __init__(self, limit):

6 self.limit = limit

7 self.n = 0

9 def __aiter__(self):

10 return self

12 async def __anext__(self):

13 if self.n >= self.limit:

14 raise StopAsyncIteration

15 await asyncio.sleep(0.1) # 异步操作

16 self.n += 1

17 return self.n

19async def main():

20 async for i in AsyncCounter(3):

21 print(f"异步迭代: {i}")

23asyncio.run(main())

25# 异步上下文管理器

26class AsyncResource:

27 async def __aenter__(self):

28 print("异步获取资源")

29 await asyncio.sleep(0.1)

30 return self

31 async def __aexit__(self, *args):

32 print("异步释放资源")

33 await asyncio.sleep(0.1)

35# async with AsyncResource() as r: ...

并发控制：信号量与超时

示例代码（可运行）

Python 0%

1# 信号量限流

2import asyncio

4async def fetch_with_limit(sem, url):

5 async with sem: # 最多 3 个并发

6 print(f"请求: {url}")

7 await asyncio.sleep(0.5)

8 return f"{url} 的结果"

10async def main():

11 sem = asyncio.Semaphore(3) # 限制并发数

12 urls = [f"url{i}" for i in range(10)]

13 results = await asyncio.gather(*[

14 fetch_with_limit(sem, url) for url in urls

15 ])

16 print(f"完成 {len(results)} 个请求")

18 # 超时控制

19 try:

20 await asyncio.wait_for(asyncio.sleep(10), timeout=1)

21 except asyncio.TimeoutError:

22 print("超时了！")

24asyncio.run(main())

🐍资深工程师经验谈

高并发 LLM 应用的架构：用 Semaphore 控制对 LLM API 的并发（避免触发限流）用 wait_for 设置超时（防止挂死）用 gather 的 return_exceptions=True 防止一个失败拖垮全部用 TaskGroup（Python 3.11+）自动管理任务生命周期流式输出用 async generator + SSE背压：用有界队列 Queue(maxsize=N) 防止生产过快消费不过来。

调度原理

事件循环的调度机制

asyncio 事件循环的核心调度逻辑：

Python

# 简化的事件循环伪代码
class SimpleEventLoop:
    def __init__(self):
        self._ready = deque()    # 就绪回调队列
        self._scheduled = []     # 定时任务(堆)

    def run_forever(self):
        while True:
            # 1. 把到期的定时任务加入就绪队列
            # 2. 执行就绪队列中的所有回调
            # 3. 如果没有就绪任务，用 selector 等待 I/O
            #    (epoll/kqueue/IOCP)
            # 4. I/O 就绪后把对应的回调加入就绪队列

# selectors 模块封装了：
# - Linux: epoll
# - macOS: kqueue
# - Windows: IOCP
# 这就是为什么 asyncio 能高效处理上万并发
# 它不是轮询，而是操作系统级的 I/O 多路复用

选择题

在 async 函数中调用 time.sleep(1) 会怎样？

选择题

限制并发请求数量应该用什么？

资深工程师加餐

底层原理 · 大厂视角 · 工程经验，点卡片展开

@decorator 写在函数定义上方，它在「模块加载、函数被定义时」就执行一次，把原函数包装成新函数；之后每次调用，走的都是包装后的版本。带参数的装饰器本质是三层嵌套（外层收参数、中层收函数、内层收调用参数）。务必用 functools.wraps 保留原函数名和文档，否则调试栈和日志里全是 wrapper。

挑战任务

并发下载器

简单+50 XP

用 asyncio + Semaphore 实现一个并发限流下载器，最多 3 个并发。

并发下载器

Python 0%

2 个测试用例

课后作业

带超时的并发

中等+30 XP

用 asyncio.gather + wait_for 实现一组任务，每个任务最多等 1 秒。

带超时的并发

Python 0%

1 个测试用例