先聊点实际的。学 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: cant start new thread。1.3 协程的思路一个人干多份活在等的时候去做别的协程的解决方案非常“取巧”既然等待 I/O 时 CPU 闲着那我就在这一小段空隙里去执行别的任务。它不像多线程那样由操作系统强行切换线程而是由程序自己决定“什么时候让出控制权”。打个比方你去医院挂号同步模式是你排队等叫号什么也干不了多线程模式是找好几个朋友一起排队协程模式则是——你在等号的时候顺便去缴费、去拿药、去打印报告全靠自己安排时间。这就是“主动让出”和“被动切换”的本质区别。Python 里协程的调度中心就是 asyncio 的事件循环它负责记住“谁在等什么”“谁可以继续跑了”。理解这层逻辑之后再去看 asyncio 的代码就会顺很多。2. asyncio 的三驾马车事件循环、协程对象与 Taskasyncio 刚接触时最容易懵的地方是名词太多事件循环、协程对象、Task、Future、awaitable……其实核心就三样其他的都是围绕这三样展开。2.1 事件循环协程的“中央调度员”事件循环Event Loop是 asyncio 的心脏。你可以把它理解成一个极度自律的排班表它维护着一个任务队列不断检查哪些任务还能继续执行、哪些任务还在等待 I/O、哪些任务已经完成了。事件循环的工作流程大致是从任务队列中取出一个可执行的任务运行它。任务执行到await时如果后面是个耗时操作就把这个任务挂起记录“它要等多久/等什么事件”。事件循环转头去执行下一个可执行任务。当某个被挂起的任务等待完成时事件循环把它重新放回可执行队列。所有任务都完成后事件循环结束。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 继承自 FutureFuture 是“未来的结果”的占位符可以给协程之外的其他代码提供异步结果。但初学者不用过分纠结 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 添加到 PATHmacOS 用户推荐用 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)是官方推荐的唯一入口函数它会做三件事创建一个新的事件循环。运行传入的协程直到它执行完毕。关闭事件循环并清理相关资源。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_exceptionsTrueresults await asyncio.gather( fail_task(), fetch_data(ok, 1), return_exceptionsTrue, ) print(results) # [ValueError(boom), ok result]这种“异常当作返回值”的处理方式在批量处理任务时非常实用。5. 新手常见问题与排查技巧asyncio 报错信息不算多但每个都很经典。我把新手最容易遇到的几个整理成速查表下面展开讲。报错/问题常见原因解决方法RuntimeError: asyncio.run() cannot be called from a running event loop在 async 函数或 Jupyter 环境里调用 asyncio.run改用 await 调用协程或在最外层再调 runcoroutine was never awaited创建了协程对象但没 await也没 create_task加上 await 或包成 task用了 async 却还是串行执行直接 await 协程没有 create_task/gather用 create_task 或 gather 并发调度异步代码里 time.sleep 导致卡死阻塞了事件循环线程替换成 await asyncio.sleeploop 已关闭/Event loop is closed老代码重复使用事件循环统一使用 asyncio.run避免手动管理 loop5.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 一次启动多个。你可以在循环里收集 Taskasync 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: fuser_{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 改一改、跑通、打印出耗时这一篇就算真正吸收了。