企业官网建设流程全解析

说实话，asyncio 这套东西，我刚接触那会儿看着网上的例子总觉得像天书——明明每个单词都认识，拼在一起就不知道程序是怎么跑起来的。后来是写了一个完整的案例，把事件循环、Task、await 这些概念全部揉进去，才真正把脑子里的线理顺了。所以这一篇我不想再讲零散的概念，直接用两个实实在在的入门案例，带你把 asyncio 从“知道”变成“会用”。如果你已经被同步代码的 I/O 等待折磨过，或者继承老项目时看到 async 关键字有点发怵，那这篇就是给你准备的。

1. asyncio 到底解决了什么问题

1.1 同步代码慢在哪儿：一个实测对比

先看一个最常见的场景：写一个脚本，要检查 100 个网站的可用状态。用同步的 requests，就是一个个来，请求发出去之后整个过程卡在那里等服务器回应，哪怕对方慢得离谱，你也只能干等着。我曾经真跑过一次，100 个 URL 里混着几个响应要 20 秒的，最终总耗时直接飙到两三分钟。

而用 asyncio 重写一遍之后，同样的 100 个请求，总耗时基本等于其中响应最慢的那一个。注意，不是 100 个请求的时间相加，而是最慢那一个的时间。百来个请求，处理完经常不到 5 秒。这个差距在批量调第三方 API、爬虫抓页面、大量小文件下载这类场景里，体验是质变。

但要注意，这里说的“慢”，指的是 I/O 密集型任务，也就是程序在等待网络、磁盘、数据库响应。如果你要做的是大量 CPU 计算，比如图像处理、复杂加密、大数据排序，那 asyncio 帮不了忙，后面我会专门聊这个。

1.2 线程、进程、协程：为什么协程更适合 I/O 密集

很多人第一反应是“有线程啊，用线程不也能并发吗？”能，但代价不一样。线程的切换是由操作系统决定的，而且每个线程都有自己独立的内存栈，开多了内存占用上去了，切换时 CPU 还要浪费时间去保存和恢复现场，又是一笔开销。

你可以把协程理解为“用户态的轻量级线程”——它不归操作系统管，而是由程序自己在合适的时候让出控制权。asyncio 的事件循环就像一个调度员，轮询每个任务：你去等这个 I/O 结果，eslint去检查下一个任务；那个 I/O 有结果了，再回来继续执行。因为切换是代码里显式的，代价极小，所以哪怕你创建成千上万个协程，也不会像开线程那样吃力。

还有一点，Python 有 GIL，多线程在纯 CPU 计算上基本是负优化，但在 I/O 等待的场景里，GIL 其实不是主要瓶颈，可线程切换的成本依旧在。协程因为压根不占内核资源，切换成本低得多，这也是它在 Python 异步编程里被推崇的主要原因。

1.3 先泼一盆冷水：什么场景别用 asyncio

我不希望你看完这篇文章，把所有代码都重写一遍。asyncio 是有适用边界的。

• 纯 CPU 密集型任务：比如视频转码、模型推理、复杂计算。这类任务没有“等待”，await完全派不上用场。
• 整个项目里只有 1-2 个请求，同步就够了，引入 asyncio 只会增加心智负担。
• 依赖的第三方库是纯同步实现，你又没法换成异步版，比如某些内部工具包，那硬用 asyncio 反而会把问题搞复杂。

最适合 asyncio 的场景，基本可以概括为一句话：很多个“等别人”的操作，彼此之间互不依赖，可以同时发出请求。这种场景，asyncio 能把你从等待的泥潭里拉出来。

2. 五个核心概念串起 asyncio 的地基

2.1 事件循环是调度中心

asyncio.run()这个函数大家应该都不陌生，它做的事其实有两件：创建事件循环，跑协程，然后关闭。事件循环就是整个异步程序的运行机制——所有协程都会注册到里面，由它决定谁先谁后，谁该等待，谁可以恢复。

用生活里的事情来类比，事件循环就像一个餐厅的前台：顾客点完餐（发起 I/O 请求），前台让顾客去休息区等（挂起协程），又去安排下一桌客人点餐（执行下一个协程）；饭菜做好了，前台喊顾客来取（I/O 完成，恢复协程）。前台不用等谁把菜吃完再做事，它只需要不停地在“谁可以点餐”“谁的菜好了”之间周旋。

2.2 协程函数调用后并不执行

这是新手最容易懵的地方。你写了一个async def fetch_data(): ...，然后在代码里调用它fetch_data()，你以为它执行了，实际上它只是创建了一个“协程对象”，里面的代码一行都没跑。我在实际教人的时候经常看到新人卡在这里，最后终端打出一行RuntimeWarning: coroutine was never awaited，这才发现自己忘了加await。

也正因为这个设计，协程的运行时机完全由你掌控。你想让它先注册，后面再 await，或者直接丢给 Task 让它自动跑，都很灵活。永远记住一句话：协程函数是一条菜谱，协程对象才是一份正在被准备的菜品。

2.3 await 只能等待“可等待对象”

在 asyncio 的世界里，await后面能跟的东西有三类：协程对象、Task、Future。泛称就是“可等待对象”。

这里我不打算光讲理论，直接给一个能跑的例子。假设我们要模拟三个请求，最快的那个先返回：

import asyncio async def request(url, delay): await asyncio.sleep(delay) return f"response from {url}" async def main(): # 三个协程，同时发起 task1 = asyncio.create_task(request("a.com", 1)) task2 = asyncio.create_task(request("b.com", 2)) task3 = asyncio.create_task(request("c.com", 0.5)) # 谁先完成，先处理谁 for coro in asyncio.as_completed([task1, task2, task3]): result = await coro print("完成:", result) if __name__ == "__main__": asyncio.run(main())

运行结果你会看到 c.com 最先打印，然后是 a.com，最后是 b.com。asyncio.as_completed就是一个“按完成顺序取结果”的迭代器，特别适合处理耗时差异大的批量请求。

2.4 Task 是对协程的包装

直接await coroutine()，等它执行完，下一个协程才会继续，这其实就是同步执行了。但很多时候我们想要的是：先把这些协程全部丢进后台，让它们同时跑，谁先完成谁先返回。那就需要asyncio.create_task()。

Task 本质上就是一个被事件循环调度执行的协程包装对象。创建 Task 之后，协程立刻开始运行（其实是进入调度队列），不需要你再显式地await它才会启动。你可以把它想象成把菜谱递给了厨师，厨师立刻开始切菜，而不需要你站在旁边盯着每一道工序。

2.5 gather、wait、as_completed 的选择

这三个函数经常放在一起比较，功能有重叠但侧重点不同。

gather适合知道要等哪些任务，并且想把结果按原顺序收集起来的情况。它的返回值是“所有结果按传入顺序排列的列表”。wait更底层，它支持等待超时时间、控制是都完成还是任一完成，还返回“已完成”“未完成”两组集合。as_completed用得最少，但在需要“每完成一个就立刻处理对应结果”的场景里非常爽快。

对照表：

3. 完整案例：把 100 个请求从 2 分钟压到 3 秒

3.1 需求与基线测试

我拿一个很常见的需求来讲：监控一批 API 接口的健康状态。假设有 100 个接口，我们要向它们发请求，统计响应状态码和耗时。同步版本我会用requests写个最朴素的 for 循环，地基就这样打。

import time import requests urls = ["https://httpbin.org/delay/1"] * 100 # 每个请求至少等 1 秒 start = time.perf_counter() for url in urls: resp = requests.get(url, timeout=5) print(resp.status_code, resp.elapsed.total_seconds()) print(f"同步耗时: {time.perf_counter() - start:.2f}s")

这段代码什么技巧都没有，但它是我们后面所有优化的基线。100 个请求，如果每个都等 1 秒以上，总耗时至少 100 秒，实际情况里还会更慢。下面我们一步步把它变成异步版本。

3.2 第一版：标准库 asyncio.to_thread 快速体验

如果你暂时不想引入第三方异步 HTTP 库，还有一条捷径：asyncio.to_thread。它可以把一个普通同步函数放到线程池里执行，再包装成协程来 await。严格来说这不是“真异步”，但在和旧代码拼接时非常实用。

import asyncio import time import requests async def fetch(url): resp = await asyncio.to_thread(requests.get, url, timeout=5) return resp.status_code async def main(): urls = ["https://httpbin.org/delay/1"] * 100 tasks = [asyncio.create_task(fetch(url)) for url in urls] results = await asyncio.gather(*tasks) print(results) if __name__ == "__main__": asyncio.run(main())

这段代码的耗时已经能降到和最后面 aiohttp 版本差不多的量级，因为阻塞操作交给线程池去扛，主循环没有被阻塞。但它也有隐患：如果 1000 个请求全部丢进线程池，线程切换成本就会重新冒头，而且你本质上还是在吃操作系统线程的资源。所以它更适合作为“改造第一步”，而不是最终方案。

3.3 第二版：aiohttp 真异步并发

真正能发挥 asyncio 全部能力的是aiohttp，这个库的 API 和 requests 很像，但所有 I/O 操作都是非阻塞的。先pip install aiohttp，然后看下面这段完整代码：

import asyncio import time import aiohttp async def fetch(session, url): async with session.get(url, timeout=5) as resp: status = resp.status text = await resp.text() return status, len(text) async def main(): urls = ["https://httpbin.org/delay/1"] * 100 async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(fetch(session, url)) for url in urls] results = await asyncio.gather(*tasks) for status, size in results: print(status, size) print(f"aiohttp 并发耗时: {time.perf_counter() - time.time_start():.2f}s") if __name__ == "__main__": asyncio.run(main())

注意我用了async with来管理连接池的会话，而不是为每个请求单独建立连接，这是性能差异非常大的一个细节。如果每次session.get都新建一个 ClientSession，开销会翻好几倍。测试下来，100 个请求在这个版本里大概 2-3 秒就能跑完，这就是真异步的效果。

不过这个例子还有个隐患：一旦某个请求异常超时，gather会直接把整个任务组炸掉。后面我们加上容错和处理。

3.4 第三版：限流、超时、重试全加上

真实项目里不能这么裸奔，至少要考虑三件事：不要让对方服务器以为你在攻击，所以要加并发限制；单个请求要么超时熔断，要么重试；异常不能把整个任务组带崩。

我直接给一个项目里可以直接拿来改的版本：

import asyncio import aiohttp urls = ["https://httpbin.org/delay/1"] * 100 async def fetch_with_retry(session, url, semaphore, retries=3): async with semaphore: # 控制同时进行的请求数 for attempt in range(retries): try: async with session.get(url, timeout=5) as resp: if resp.status < 400: return resp.status, await resp.text() else: raise aiohttp.ClientError(f"http {resp.status}") except (aiohttp.ClientError, asyncio.TimeoutError) as exc: if attempt == retries - 1: return None, str(exc) await asyncio.sleep(0.5 * (attempt + 1)) # 退避 return None, "failed" async def main(): semaphore = asyncio.Semaphore(20) # 最多 20 个并发 async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(fetch_with_retry(session, url, semaphore)) for url in urls] results = await asyncio.gather(*tasks) success = sum(1 for status, _ in results if status) print(f"成功 {success}/{len(urls)}") if __name__ == "__main__": asyncio.run(main())

这套代码的核心是Semaphore(20)，它就像一个并发闸门，保证同时只有 20 个请求在飞。重试时用了指数退避的简化版本，第一次失败等 0.5 秒，第二次等 1 秒，第三次彻底放弃。整体逻辑已经和真实生产代码非常接近了。

4. 进阶三板斧：并发限制、超时与取消

4.1 Semaphore 限流的底层逻辑

信号量这个东西，说穿了就是一个计数器。你创建asyncio.Semaphore(20)，它初始值是 20。每次进入async with semaphore时，如果计数大于 0 就把计数减 1，放你过去；如果已经是 0，你就得在门口排队，等前面的人出来时把计数加回 1，再放你进去。

为什么不能直接发 1000 个并发的请求？不是因为 asyncio 扛不住，而是对端服务器扛不住。很多服务端对单 IP 并发连接数有限制，可能 50 个并发就把你拒之门外了。从你自己的角度，连接池的连接数也有上限，ClientSession 默认连接数在 100 左右，超过之后照样排队。所以在学 asyncio 的同时把 Semaphore 用熟，是实际项目和“玩具代码”的分水岭之一。

4.2 wait_for 超时的坑与正确姿势

asyncio 的wait_for用起来很简单：result = await asyncio.wait_for(coro, timeout=10)。10 秒内没完成就抛出asyncio.TimeoutError。听起来很美好，但有个坑：超时后任务不会自动“消失”，它会被取消（cancel）并等待其清理完成。如果你的协程里有收尾工作，比如释放连接、写日志，取消信号什么时候到，是由事件循环决定的，不是立刻终止。

还有就是，不要在wait_for里传一个简单的协程同时又塞进任务列表，不然它会变成两个独立的对象执行。我有一次写了一批请求，用create_task把协程丢进列表，然后又对同一个协程对象wait_for(..., timeout=3)，结果任务执行了两遍，接口被重复调了。根源就在于协程对象只能被 await 一次，你传进去的引用还是原来那个。

4.3 任务取消与 shield 的取舍

任务取消的机制很优雅，但也很容易踩坑。task.cancel()会在协程内抛一个asyncio.CancelledError，协程可以选择在这个异常处正常退出，也可以用try/finally做清理。关键点是：如果你捕获了CancelledError又不重新抛出，事件循环会认为任务已经被取消了，但实际它在后台“苟活”着，容易造成逻辑错乱。

shield()则相反，它保护一个协程不被取消。默认行为是取消信号直接打到 shield 上，底层协程继续跑。但注意，如果底层协程自己不去理会取消，那 shield 也会跟着完蛋。我见过不少人在超时场景里加了 shield，以为万事大吉，结果底层用的是同步阻塞库，取消信号根本不生效。所以要在“真异步”的前提下讨论取消，混入同步阻塞逻辑后这些都是空谈。

4.4 三种结果收集方式对比

前面提到过 gather、wait、as_completed，这里我再多说一层。我在实际使用中的习惯是这样的：

• 如果任务之间没有依赖，但要等所有结果，优先gather，因为它返回结果列表的顺序稳定，方便和 URL 列表一一对应。
• 如果某个任务失败不能影响整体，那就给 gather 加return_exceptions=True，让它把异常封装进返回列表，而不是往上抛。
• 如果第一个成功的结果就能继续推进流程，比如搜索接口并发多个服务商，谁快用谁，那就用asyncio.wait配合FIRST_COMPLETED，拿到第一个就取消其余。

这里给一个FIRST_COMPLETED的简要示例：

import asyncio async def service_a(): await asyncio.sleep(3) return "a" async def service_b(): await asyncio.sleep(1) return "b" async def main(): task_a = asyncio.create_task(service_a()) task_b = asyncio.create_task(service_b()) done, pending = await asyncio.wait( {task_a, task_b}, return_when=asyncio.FIRST_COMPLETED, ) for task in pending: task.cancel() # 赢了就不用等慢的那个了 print(done.pop().result()) if __name__ == "__main__": asyncio.run(main())

这种竞速模式在微服务调用、多路数据源选路时特别好用。

5. 真实项目里最常见的 5 个坑

5.1 RuntimeError：Event loop is closed

你很可能在 Jupyter Notebook 里遇到过这个报错。原因很简单：Jupyter 自己维护了一个事件循环，而你在同一个进程里调用了两次asyncio.run()，第二次调用时旧事件循环已经被关闭了，新事件循环又想用同一个东西，直接冲突。

解决办法也简单：在 Jupyter 里不要用asyncio.run()，用await配合await asyncio.create_task(...)或者直接在 notebook 的顶层执行asyncio.run前面的代码块。另一个常见场景是 Flask/Django 这种同步框架里直接用 asyncio，往已有的事件循环里塞任务时也会遇到类似问题。这时候用asyncio.set_event_loop(asyncio.new_event_loop())换个新循环，或者改用nest_asyncio（但这是补丁式方案，心里要有数）。

5.2 协程里混用同步库，导致“假异步”

这是最大的一个坑。你写了一堆async def，看着井井有条，但里面突然有一行requests.get()，这行同步请求一旦发出去，整个事件循环就被阻塞住了，其他协程全在坑里排队。你写的异步代码瞬间变成一张华丽的外皮。

我排查过很多类似的问题，表现是：程序“异步改造”后耗时几乎没降。第一反应不是怀疑并发逻辑，而是找找有没有同步阻塞调用混在里面。解决方案有三个：换 aiohttp/httpx 这样的异步客户端；或者用asyncio.to_thread把同步函数丢进线程池；再或者，如果你是在 FastAPI 等框架中，可以直接用run_in_executor把阻塞任务隔离到独立线程。

5.3 Task exception was never retrieved

这个报错的含义是：某个 Task 里抛了异常，但没有任何地方接收这个异常。事件循环检测到后，会在垃圾回收时打出一条警告。示例：

import asyncio async def bad(): raise ValueError("boom") async def main(): task = asyncio.create_task(bad()) # 没有 await，也没有加异常处理 await asyncio.sleep(0.1) asyncio.run(main())

运行后你就会看到Task exception was never retrieved的警告。解决方法是所有 Task 都尽量有归属：要么收集进列表后用gather或者wait统一处理；要么给 Task 添加add_done_callback去检查异常；要么在每个协程内部就做好 try/except，把异常吞掉并记录日志。总之，不要让任何异常悬空。

5.4 回调地狱与 async/await 之间的关系

老牌异步框架 Twisted 和 asyncio 的早期风格，都鼓励用回调：请求完成后调这个函数，失败后调那个函数。回调一嵌套，代码就变成了“箭头形”，阅读和排查都很痛苦。asyncio 里虽然也有loop.add_reader这种偏底层的回调机制，但我们现在写业务代码，我的建议是能用 await 就不要用回调。

这两个风格怎么选？哪怕是协程之间的链式调用，也优先用await把数据流写清楚，而不是把一个函数的返回值作为参数传给另一个回调。唯一推荐回调的场景是在事件循环的底层 API、或者在自定义事件循环集成时，普通业务代码几乎用不到。

5.5 排查利器：debug 模式与慢回调日志

写 asyncio 程序出 bug 时，靠 print 大法效率太低了。建议学会打开 asyncio 的调试模式：在代码开头设置loop.set_debug(True)，或者在环境变量里设置PYTHONASYNCIODEBUG=1。开启之后，事件循环会记录每个回调的执行时长，如果发现某个协程执行时间超过默认阈值（通常是 100ms），日志里会标出“slow callback”。

排查思路一般是：先开 debug 模式，看是哪个协程耗时异常；再沿着日志定位是 I/O 等待还是在做 CPU 计算；然后看是不是有同步阻塞库混入。我曾通过这种办法发现一个第三方 SDK 内部用的是同步requests，藏得特别深，不开 debug 根本发现不了。

6. 走出 Python：协程思想的跨语言迁移

6.1 Kotlin 协程和 Flow 是什么

虽然标题是 Python 的 asyncio，但“协程”这套心智模型不止一个语言在用。现在 Android 开发里很火的 Kotlin 协程就是典型：suspend fun相当于 Python 里的async def，withContext用来切换线程，CoroutineScope相当于事件循环的作用域。而 Flow 则是 Kotlin 做异步数据流的一套方案，类似 Python 里的异步生成器。

很多 Python 程序员切过去没什么障碍，因为两者的本质是一样的：用顺序写的方式表达异步流程，把线程切换交给框架而不是程序员。

6.2 从 asyncio 到 Flow：它们想解决同一件事

如果你理解 asyncio 里“await 一个 I/O 操作时让出控制权”的逻辑，那你理解 Kotlin 协程也不会太难。Flow 在它的基础上多了“上游发射数据、下游收集数据”的模型，收集过程中一旦遇到网络请求这种 I/O，照样会挂起并在合适的时机恢复。

这两种语言里的异步观念完全同源，我甚至觉得多学一门语言的异步实现，能反过来加深对另一门语言的理解。当然，这篇的篇幅有限，我的建议是先把 Python 这套玩熟，再跨过去看 Flow，你会发现很多概念可以直接平移。

6.3 用同理心去学任何语言的异步

最后说点题外话。异步编程的难点不在语法，而在于心智模型的切换：你不能再按“第一行执行完再执行第二行”的老思路去读代码，而要时刻去想“这里挂起了吗？挂起多久？谁在等它？”。一旦接受了这个设定，你学任何语言都有了一座桥。反过来，如果只是背 API，不理解事件循环和任务调度，到了下一个语言又会回到从零开始的状态。

结尾

在我实际带新人的过程中，协程这一块最常出现的问题是“看得懂例子，不敢写逻辑”。我的建议一直很简单：找一个自己手头正在做的同步小脚本，比如批量检查接口、批量下文件、批量发通知，照着这篇的思路逐步改造成 asyncio 版本，跑通了再继续往里加限流、超时、重试和取消。当你把一个真实脚本改造成功的那一刻，对事件循环、Task、Semaphore 这些概念的体感会和看完十篇教程都不同。

如果真要说一个最容易被人忽略的小技巧，那就是：写 asyncio 代码时，每个协程内部都要有独立的异常处理，不要把风险都抛给最外面的gather。我见过太多线上事故都是从“某个请求异常导致整批任务都退出”开始的，你花半小时给所有 create_task 加上兜底日志，未来能省下几个通宵。