Python协程与asyncio入门：从事件循环到并发编程实战

先聊点实际的。学 Python 绕不过并发，而 Python 并发绕不过协程。只要你写过爬虫、做过接口轮询、处理过大量 I/O 操作，一定体会过“程序卡在等待上”的那种无力感。比如你用 requests 下载几十个文件，一个请求没回来，后面全堵着；换成多线程，又要处理锁、队列、线程安全，代码复杂度直线上升。这时候 asyncio 就是那个能让你用单线程，却把 I/O 等待时间“偷”回来用的东西。

这篇文章是“python3 从入门到精通”系列里专门讲 asyncio 的第一篇，我会从最底层的概念开始，把事件循环、协程对象、Task、await 这些容易绕晕的名词一个个拆开讲，再带你写第一个真正“并发”的异步程序。适合已经有 Python 基础、但第一次接触协程的读者，目标是让你读完就能动手写异步代码，并且知道自己写的每一行在干什么。

1. 先搞懂：协程到底在解决什么问题

很多教程上来就甩 async/await 语法，但如果不理解它解决什么问题，写出来的代码大概率是“代码会跑，逻辑不对”。所以我习惯先讲清楚痛点。

1.1 同步代码为什么慢：一次请求堵住整条路

先看一段最常见的同步爬虫代码思路：

import time def fetch_one(url): print(f"开始请求: {url}") time.sleep(2) # 模拟网络请求耗时 print(f"请求完成: {url}") def main(): for url in ["a.com", "b.com", "c.com"]: fetch_one(url) start = time.time() main() print(f"总耗时: {time.time() - start:.2f}s")

运行结果很直观：3 个请求，每个 2 秒，总耗时 6 秒。问题出在哪？time.sleep(2)模拟的是网络等待，这段时间 CPU 其实什么都没干，但它就是干等着，不让后面的代码执行。这种“一个人干活，等别人交差才能继续”的模式，在 I/O 密集场景下浪费极其严重。

爬虫、文件读写、数据库查询、调用第三方 API，本质上都是这种“发出请求 — 等待响应 — 拿到数据”的模式。等待时间往往比实际计算时间长好几个数量级，而同步代码把等待时间全部浪费掉了。

1.2 多线程不是万能的：线程切换也是成本

有人会说：那用多线程不就行了？对，多线程确实能解决一部分问题。Python 的threading也可以同时发起多个请求，但多线程有几个绕不开的麻烦。

首先是 GIL。Python 的全局解释器锁决定了同一时刻只能有一个线程执行 Python 字节码，所以多线程在 CPU 密集型任务上几乎没有任何加速效果。不过对于 I/O 密集型任务，由于线程在等待 I/O 时会释放 GIL，多线程确实有效果，这也是很多爬虫用多线程的原因。

但多线程真正的痛点在于复杂性：多个线程同时读写共享变量需要加锁，加锁又可能引入死锁、竞态条件。线程的创建和切换也有开销，几千个线程会让系统难堪重负。我曾经处理过一个爬虫项目，开 500 个线程抓数据，结果对方服务器直接把我 IP 封了，自己的程序也频繁报thread.error: can't start new thread。

1.3 协程的思路：一个人干多份活，在等的时候去做别的

协程的解决方案非常“取巧”：既然等待 I/O 时 CPU 闲着，那我就在这一小段空隙里去执行别的任务。它不像多线程那样由操作系统强行切换线程，而是由程序自己决定“什么时候让出控制权”。

打个比方：你去医院挂号，同步模式是你排队等叫号，什么也干不了；多线程模式是找好几个朋友一起排队；协程模式则是——你在等号的时候，顺便去缴费、去拿药、去打印报告，全靠自己安排时间。这就是“主动让出”和“被动切换”的本质区别。

Python 里协程的调度中心就是 asyncio 的事件循环，它负责记住“谁在等什么”“谁可以继续跑了”。理解这层逻辑之后，再去看 asyncio 的代码，就会顺很多。

2. asyncio 的三驾马车：事件循环、协程对象与 Task

asyncio 刚接触时最容易懵的地方是名词太多：事件循环、协程对象、Task、Future、awaitable……其实核心就三样，其他的都是围绕这三样展开。

2.1 事件循环：协程的“中央调度员”

事件循环（Event Loop）是 asyncio 的心脏。你可以把它理解成一个极度自律的排班表：它维护着一个任务队列，不断检查哪些任务还能继续执行、哪些任务还在等待 I/O、哪些任务已经完成了。

事件循环的工作流程大致是：

1. 从任务队列中取出一个可执行的任务，运行它。
2. 任务执行到await时，如果后面是个耗时操作，就把这个任务挂起，记录“它要等多久/等什么事件”。
3. 事件循环转头去执行下一个可执行任务。
4. 当某个被挂起的任务等待完成时，事件循环把它重新放回可执行队列。
5. 所有任务都完成后，事件循环结束。

Python 3.10 之前的写法需要手动获取和关闭事件循环，3.10 以后官方推荐直接用asyncio.run()，它内部会帮你创建事件循环、运行协程、关闭事件循环，一步到位。所以我建议新代码统一用asyncio.run()，别再去碰老式的loop = asyncio.get_event_loop()那套写法了。

2.2 async def 与 await：协程的“灵魂语法”

async def定义一个协程函数，调用它会返回一个协程对象。注意这个细节：它不会执行函数体，只是返回一个对象。

async def hello(): print("Hello, asyncio!") coro = hello() print(coro) # <coroutine object hello at 0x...>

这时候coro就是一个协程对象，但它还没真正运行。要让它的代码执行起来，要么直接await coro，要么把它交给事件循环。

await是协程里的“暂停点”。await asyncio.sleep(1)的意思是：执行到这里，我告诉事件循环“我要等 1 秒，这 1 秒你先去干别的，时间到了再叫我回来”。这就是协程能并发执行的底层机制——没有await的主动让出，就没有协程的并发调度。

注意：在async def函数里，可以写await；但在普通def函数里写await会直接语法报错。如果你在普通函数里想调用异步函数，不能用await，得用asyncio.run(coro)或者在另一个协程里去await它。

2.3 Task：真正进入事件循环的“任务凭证”

协程对象本身不够“事件循环”用，它需要一个“壳”把它包装成可调度的任务，这个壳就是 Task。Task 可以理解成协程的高级版本：它额外记录了运行状态、结果、异常等信息，并且可以提前在“后台”启动，不用等 someoneawait它才执行。

asyncio.create_task()是把协程包装成 Task 的推荐方式：

async def say_hello(): await asyncio.sleep(1) print("Hello") task = asyncio.create_task(say_hello())

这行代码一执行，say_hello()的协程就已经被注册到事件循环里了，事件循环会在合适的时机运行它。你不需要立刻await它，可以在里面插入其他代码，最后再统一await收集结果。这正是并发的基础。

我还想提一下 Future。Task 继承自 Future，Future 是“未来的结果”的占位符，可以给协程之外的其他代码提供异步结果。但初学者不用过分纠结 Future 的内部实现，先把 Task 用明白就够了。

3. 第一个异步程序：从 3 行代码跑到会调度的级别

看再多概念不如动手写一个。下面我带你把第一个 asyncio 程序跑起来，并且跑一个真正的“并发”效果出来。

3.1 环境准备：Python 版本建议 3.7 以上

asyncio 在 Python 3.4 引入，但早期 API 很不完善。asyncio.run()在 3.7 才加入，asyncio.create_task()也是 3.7 加入的。所以我建议直接用 Python 3.8 以上版本，3.11 和 3.12 的表现会更好，错误提示也更友好。

终端确认版本：

python3 --version

如果版本太低，就先去 Python 官网下载安装新版本。Windows 用户注意把 Python 添加到 PATH，macOS 用户推荐用 Homebrew 安装，Linux 用户可以用包管理器。

3.2 第一个异步程序：两个协程“看似同时”执行

先写一个最简单、能看懂的：

import asyncio async def say_hello(): print("hello start") await asyncio.sleep(1) print("hello end") async def main(): await say_hello() await say_hello() asyncio.run(main())

这段代码输出结果是：

hello start hello end hello start hello end

总耗时约 2 秒。因为这里await say_hello()是“等第一个跑完，再跑第二个”，属于顺序执行，还不是并发。很多新手在这一步就懵了：我用的明明是 async/await，为什么还是串行的？

因为这里你只用了await直接等待协程对象，没有把它包装成 Task 让它们在后台并发。

3.3 同步版本对比：知道它比什么快

为了感受差异，先看同步版本：

import time def sync_say(): print("hello start") time.sleep(1) print("hello end") start = time.time() sync_say() sync_say() print(f"同步耗时: {time.time() - start:.2f}s")

输出大致是 2 秒。如果换成 asyncio 版本，但代码写成了上面 3.2 那样，依然是 2 秒。所以“用了 asyncio”不等于“快了”。

3.4 让任务真正并发：create_task 的典型姿势

正确姿势是用asyncio.create_task()把协程提前放进事件循环：

import asyncio import time async def say_hello(name): print(f"{name} start") await asyncio.sleep(1) print(f"{name} end") async def main(): task1 = asyncio.create_task(say_hello("task1")) task2 = asyncio.create_task(say_hello("task2")) await task1 await task2 start = time.time() asyncio.run(main()) print(f"异步耗时: {time.time() - start:.2f}s")

运行结果：

task1 start task2 start task1 end task2 end 异步耗时: 1.00s

注意输出的顺序：task1 start和task2 start是连续打印的，两个协程分别在等待 1 秒的期间都“启动”了。总耗时约 1 秒，而不是 2 秒。这就是异步 I/O 的核心效果：把两个 1 秒的等待时间重叠了。

这里我强烈建议你自己敲一遍这段代码，然后故意把await task1改成await say_hello("task1")，再看运行结果，你就能深刻理解create_task的差异了。

4. asyncio 高频 API 与参数选择

asyncio 的 API 不算多，但每个都有适用场景。我挑了最常用的几个，结合踩过的坑一起讲。

4.1 asyncio.run：新时代的入口函数

asyncio.run(coro)是官方推荐的唯一入口函数，它会做三件事：

1. 创建一个新的事件循环。
2. 运行传入的协程，直到它执行完毕。
3. 关闭事件循环，并清理相关资源。

import asyncio async def main(): print("run me") asyncio.run(main())

有一个常见误区是 “在已经运行的事件循环里再调asyncio.run()”。比如你在 Jupyter Notebook 里使用 asyncio，或者在一个async函数里又写了asyncio.run(some_coro())，就会报错：RuntimeError: asyncio.run() cannot be called from a running event loop。因为一个线程里同时只允许有一个事件循环在运行。

4.2 await asyncio.sleep 与 time.sleep 的大坑

在协程里千万不要用time.sleep()。它会阻塞整个线程，导致事件循环卡住，其他任务全部无法执行。

import asyncio import time async def bad_demo(): print("start") time.sleep(2) # 会阻塞事件循环！ print("end") async def main(): task1 = asyncio.create_task(bad_demo()) task2 = asyncio.create_task(bad_demo()) await task1 await task2

这段代码总耗时约 4 秒，因为time.sleep(2)把整个事件循环线程都卡住了，task2在task1睡完之前根本没机会执行。换成await asyncio.sleep(2)才是正解。

4.3 gather、wait、create_task 怎么选

这三个 API 都能同时“跑”多个协程，但有区别。

asyncio.gather用的最多，它可以同时收集多个协程/Task 的结果：

async def fetch_data(name, delay): await asyncio.sleep(delay) return f"{name} result" async def main(): results = await asyncio.gather( fetch_data("a", 1), fetch_data("b", 2), ) print(results)

gather有个特点是“一荣俱荣，一损俱损”，如果其中某个协程抛异常，默认会直接向外抛出，其他还没执行完的任务行为取决于return_exceptions参数。我建议在需要同时启动多个任务并收集结果时，优先用gather。

asyncio.wait更底层一些，它接收一组 Task/Future，返回完成和未完成两个集合，适合需要自定义“等多久、等几个”的场景，比如“只要有一个返回我就继续”。

create_task则是把一个协程变成 Task，灵活度最高，但需要你自己管理句柄，然后await task或者task.result()获取结果。

我通常的选择标准：要结果、数量固定、希望一起跑，用gather；要精细控制超时或“部分完成即可”，用wait；要在执行过程中穿插业务逻辑、动态添加任务，用create_task。

4.4 异常处理的顺序问题

异步并发里最容易忽略的就是异常。如果gather里某个协程抛了异常，程序会直接跳出来，但其他协程产出的结果可能已经丢了。

async def fail_task(): raise ValueError("boom") async def main(): try: results = await asyncio.gather( fail_task(), fetch_data("ok", 1), ) print(results) except ValueError as e: print("捕获异常:", e)

这样写，fetch_data("ok", 1)的结果可能拿不到，因为gather在异常发生时已经中断了。如果想要“其他结果不受影响”，可以设置return_exceptions=True：

results = await asyncio.gather( fail_task(), fetch_data("ok", 1), return_exceptions=True, ) print(results) # [ValueError('boom'), 'ok result']

这种“异常当作返回值”的处理方式，在批量处理任务时非常实用。

5. 新手常见问题与排查技巧

asyncio 报错信息不算多，但每个都很经典。我把新手最容易遇到的几个整理成速查表，下面展开讲。

5.1 RuntimeError: asyncio.run() cannot be called from a running event loop

这个错最常见于 Jupyter Notebook、 IPython 交互环境，或者扩展现有 async 框架（比如 FastAPI 里再写 asyncio.run）。原因是这些环境本身已经有事件循环在运行，你再创建一个新的当然不允许。

如果你是脚本程序，把asyncio.run()的调用放到最顶层的main里就对了。如果是在 Jupyter 里，别用asyncio.run()，直接把协程写成await在 cell 中执行。还有一种做法是用nest_asyncio库，它能修补底层让循环嵌套运行，但我不建议在生产代码里用，只适合快速调试。

5.2 coroutine was never awaited

这是很多刚接触 asyncio 的人第一步就会遇到。原因特别简单：你写了一个async def函数，然后像普通函数一样调用它，但它返回的是协程对象，不是结果。

async def get_data(): return 42 data = get_data() # 得到协程对象，不是 42

更隐蔽的情况是忘记await。比如在列表推导里：

results = [get_data() for _ in range(3)] # 三个协程对象，永远不执行

正确写法：

results = [await get_data() for _ in range(3)]

或者用asyncio.gather(*(get_data() for _ in range(3)))并发执行。

5.3 本以为并发，结果还是顺序执行

这种问题复盘起来特别有意思。常见代码是：

async def main(): for i in range(3): await fetch_data(i)

你用了 async/await，但await fetch_data(i)会等待当前任务完成，才进入下一轮循环，这不是并发。本质原因是你没有创建 Task，也没有用 gather 一次启动多个。

你可以在循环里收集 Task：

async def main(): tasks = [] for i in range(3): tasks.append(asyncio.create_task(fetch_data(i))) await asyncio.gather(*tasks)

或者直接一行 gather 加生成器。搭配time.time()测总耗时，效果立竿见影。

5.4 在异步函数里用了 time.sleep，整个程序卡死

前面讲过这个坑，我再补充一个观察技巧：如果程序打印完一堆 start 然后沉默很久，再一次性打印一堆 end，多半就是引入了同步阻塞调用。排查时可以看自己是用了time.sleep、requests.get，还是别的同步 I/O 库。

记住：只要想使用 asyncio 的并发能力，所有耗时的 I/O 操作都要尽量使用异步版本。比如网络请求用aiohttp、httpx.AsyncClient，文件操作用aiofiles，数据库访问用支持异步的驱动。同步库不会自动“异步化”。

5.5 事件循环突然跑了很久不结束

有次我写异步脚本，主协程跑完了程序却迟迟不退出。后来发现是有个 Task 没有await，事件循环在等它结束。排查方法是检查是否有“游离”的 Task，或者用asyncio.all_tasks()打印当前所有未完成任务。遇到这种情况，批量创建任务后最好统一await asyncio.gather(*tasks)，避免任务“跑飞”。

6. 实战小场景：用 asyncio 同时请求多个接口

概念讲完，我带你串一个完整的小项目：并发请求三个假接口，分别拿到数据后再做合并处理。这段代码可以直接改造成真实爬虫或 API 调用。

import asyncio import time # 模拟异步网络请求 async def fetch_user(id: int): await asyncio.sleep(1) # 模拟网络延迟 return {"user_id": id, "name": f"user_{id}"} async def fetch_orders(user_id: int): await asyncio.sleep(1.5) return {"user_id": user_id, "orders": [101, 102, 103]} async def fetch_profile(user_id: int): await asyncio.sleep(0.8) return {"user_id": user_id, "vip": True} async def main(): start = time.time() # 三个请求并发执行 user, orders, profile = await asyncio.gather( fetch_user(1), fetch_orders(1), fetch_profile(1), ) # 合并数据 result = {**user, **orders, **profile} print(result) print(f"并发总耗时: {time.time() - start:.2f}s") asyncio.run(main())

这段代码里，fetch_user等 1 秒，fetch_orders等 1.5 秒，fetch_profile等 0.8 秒。如果用同步写法，总耗时是 1 + 1.5 + 0.8 = 3.3 秒。用 asyncio 之后，它们并发执行，总耗时由最慢的那个决定，也就是约 1.5 秒。数据到达顺序可能不一样，但gather会保证返回结果顺序和传入顺序一致，这一点在实际业务里很关键。

我自己的经验是，光理解“并发”概念还是不够的，一定要亲手计时，看到 3.3s 变 1.5s 才有体感。有了这个手感，再去看 aiohttp 爬虫、FastAPI 异步接口、消息队列消费逻辑，都会顺很多。

这篇是 asyncio 系列的第一篇，重点在搭建心智模型和跑通基本流程。下一期可以聊更进阶的东西：锁与信号量、队列、超时控制、Task 取消，以及如何用 async 的思维去设计一个并发爬虫。先把手上的 demo 改一改、跑通、打印出耗时，这一篇就算真正吸收了。