企业官网建设流程全解析

很多人第一次接触 Python 网络并发编程，首先听到的一定是线程和进程。但等到真正做爬虫、做接口聚合、做代理转发这类大量 IO 等待的任务时，才会意识到协程才是赢家。尤其当你已经学会用线程池解决问题，却仍然被内存占用、GIL 锁、线程切换开销搞得很头疼的时候，把协程拿出来重新审视一遍，是一个非常值得花时间做的事。这篇文章就是针对协程这一块，把网络并发编程中最实用的部分拆开讲清楚，包括 async/await 背后的执行模型、asyncio 事件循环的工作方式，以及一个可以立刻上手的 aiohttp 并发抓取案例，同时把我个人踩过的坑和一些性能边界判断的经验一并整理出来。

这不是一篇介绍"什么是协程"的入门科普。我已经默认你写过装饰器、知道函数是怎么调用的、也大概听过生成器里的 yield 有暂停功能。如果你正处在"能写同步代码但并发写不好"的阶段，这篇文章应该是最适合你的那种进阶实操文。

1. 为什么网络并发场景里协程是最终答案

在还没有协程这一套普及之前，大家遇到网络 IO 密集型的任务，第一反应都是用线程。爬虫要抓100个页面，那就开100个线程。每个线程里用 requests 发请求，然后等待返回。这套逻辑看起来没什么问题，但放到真实环境里很容易被拖垮。

1.1 IO 密集任务的本质：CPU 大部分时间都在等

网络请求这个动作，本质上是把数据包发出去，然后等对方服务器处理、再等数据包传回来。在你发出请求到拿到响应的这段空档，你的 CPU 几乎什么都没干。如果用一个简单的比喻：你叫了一份外卖，等外卖的时间里，你在电脑前盯着订单页面发呆，动不了别的。线程模式就是这样干的，一个线程对应一个请求，这个请求在等外卖，整个线程就被吊住了。

这里有一个非常关键的事实：Python 的线程并不便宜。每个线程要占用自己的栈空间，线程之间切换需要操作系统参与，而且因为 GIL 的存在，即便你有 8 个核，同一时刻跑 Python 字节码的线程就只有一个。写网络请求代码的每一行 Python 代码都要受 GIL 约束，虽然底层 socket 的收发操作会释放 GIL，但线程调度、上下文切换、锁竞争这些事情全都是额外开销。

所以你开 200 个线程去抓页面，实际表现是：线程数量一多，系统上下文切换开销骤增，内存也节节攀升，而 CPU 本身却没有多少真正干活的时间。这是几乎所有用 requests 多线程爬虫的同学们都会碰到的问题。

1.2 协程的核心优势：单线程内的"边等边干"

协程做的事情，说白了就是在单线程里，把一个函数在执行到某个 IO 等待点的时候暂停，去执行另一个函数，等前者的 IO 完成了再回来继续。整个过程只有一个线程，不存在线程上下文切换，也不涉及多线程之间的资源共享和锁问题。因为只有一个线程，所以根本不需要 GIL 的保护来保证变量安全，这一点在写并发代码的时候会轻松非常多。

再拿外卖比喻：协程模式不是你盯着外卖订单页面发呆，而是你下单之后去看书、去写邮件、去开会，外卖到了之后你再去取。一个人的时间利用率被拉满了。

在 Python 里，这个"等外卖"的动作就是await。await后面只能跟一个可等待对象，通常是一个协程。当你await一个网络请求时，当前这个协程就暂停，事件循环看到这个协程在等 IO，就立刻去调度别的协程。这个"暂停并让出控制权"和生成器的yield很像，但语义上更明确：它专门用于挂起等待结果，而不是产出数据。

1.3 我用一组真实数据说服自己

不亲自测一次，永远不知道协程的威力。我做过一个很简单的测试，用百度、 GitHub、 和几个公开 API 一共 30 个 URL，分别用三种方式跑：

• 同步 requests：每个请求约 0.4 秒到 1 秒不等，30 个全部跑完大概用了 18 秒。
• 线程池（ThreadPoolExecutor，最大线程数 20）：跑完大概用了 3 秒。
• asyncio + aiohttp（信号量限制 10）：跑完大概也是 3 秒，但线程池版本的内存占用是协程版本的 3 倍左右。

协程版本跑完 30 个请求的收益已经很明显了，如果请求数增加到 500 个，线程池版本基本就开始出现连接超时、内存飙高，而协程版本只要控制好并发数，表现依旧稳定。这就是为什么我后来把所有网络 IO 密集型的代码都优先考虑协程。

2. async/await 语法背后那台隐形发动机

协程的语法本身很简单，一个async def定义协程函数，函数调用返回一个协程对象，然后用asyncio.run()或者事件循环去驱动。真正难理解的是它背后的事件循环机制。不少人写协程代码，写着写着就卡在"为什么我的协程不执行""为什么 await 之后代码顺序不对"这些问题上，基本都是因为没有理解事件循环是怎么调度的。

2.1 事件循环：协程的调度中枢

asyncio.run(main())做的事情，说到底就是创建了一个事件循环，把main()这个协程扔进去跑，跑完后关闭循环。这个事件循环，你可以把它看成是一个"待办事项队列 + 完成回调管理器"。

当一个协程执行到await的时候，它就会告诉事件循环："我现在要等一个东西，你能不能先去跑别的协程？" 事件循环说好，然后把当前协程挂起，去事件队列里看还有谁可以跑。当 await 的那个 IO 操作完成时，通常是一个 socket 收到数据或者文件读出数据，事件循环会收到操作系统的事件通知，再把对应的协程唤醒，塞回"可执行队列"里。

所以整个事件循环的核心结构，就是一个不断重复的循环：从可执行队列取一个协程，执行到它遇到 await 挂起，然后取下一个。没有待执行任务时就 sleep 等待 IO 事件。这一切都在同一个线程里面完成，看起来像在同时做很多事，实际上是在非常高效地轮流做事。

2.2 一个极小的调度案例

我推荐用下面这个例子理解事件循环调度顺序：

import asyncio async def task_a(): print("A start") await asyncio.sleep(1) print("A end") async def task_b(): print("B start") await asyncio.sleep(0.5) print("B end") async def main(): taskA = asyncio.create_task(task_a()) taskB = asyncio.create_task(task_b()) print("main waiting...") await taskA await taskB asyncio.run(main())

输出顺序是：

main waiting... A start B start （等待 0.5 秒） B end （再等 0.5 秒） A end

注意看，B start出现在main waiting...之后，而且A start和B start是紧接着输出的。这在同步代码里不可能发生，因为在同步代码中，A 的sleep(1)会阻塞整个线程。但在协程里，asyncio.sleep(1)并不是真的让这个线程睡 1 秒，而是告诉事件循环"1 秒后唤醒我"，这 1 秒里事件循环会去执行 B。这就是"并发等待"的基本模型。

2.3 为什么 async def 函数调用不立即执行

这是初学者最容易困惑的一点。你写了一个async def函数，然后调用它，Python 并不会执行函数体，而是返回一个 coroutine 对象。只有把这个协程交给事件循环执行，函数体里直到第一个await之前的代码才会真正跑起来。

打个比方：你定义了一个任务，相当于写好了一份菜谱。直接调用函数，只是拿到了这份菜谱，并没有开始做菜。asyncio.run()相当于进了厨房开始照着菜谱做菜，遇到await就是在等某个食材炖熟，这时候厨师（事件循环）会去做另一道菜的准备工作。

这个理解很重要，因为经常有人写出这样的代码：

async def fetch(url): ... # 错误用法：没有把协程交给事件循环 fetch("https://example.com") # 正确用法 asyncio.run(fetch("https://example.com"))

直接调用fetch()后，你会发现什么都不会发生，连一个 print 都不会出现。只有把它们包进事件循环里，才能跑起来。如果你在一个协程里想要"并发"跑多个协程，则要用asyncio.gather()或者asyncio.create_task()，而不是简单地写await fetch(a); await fetch(b)，因为那样会变成串行。

3. 实战：用 aiohttp 把单线程爬虫提速 10 倍

说了这么多原理，直接上一个可以跑的项目。假设我们要抓取一个大型站点的 100 个详情页，最简单的同步版本是循环发 requests 请求。协程版本的思路完全一样，只是把 requests 换成支持异步的 aiohttp，用协程发起并发请求，然后用信号量限制最大并发数，避免把对方服务器打崩或者把自己这边的连接数撑爆。

3.1 环境准备与依赖安装

建议直接用 Python 3.10 及以上版本，低版本虽然也能跑，但 3.10 之后事件循环的 API 更稳定，官方也推荐统一走 asyncio.run()。你需要安装 aiohttp：

pip install aiohttp

如果要解析 HTML，建议配合 BeautifulSoup4 或 lxml：

pip install beautifulsoup4 lxml

我建议在一个独立的虚拟环境里装，不要直接污染系统 Python 环境。顺手把文件命名为crawl_async.py，这样后面调试也方便。

3.2 基础版并发爬虫示例

一个最基础的 aiohttp 并发抓取长这样：

import asyncio import aiohttp async def fetch(session, url): async with session.get(url, timeout=10) as resp: resp.raise_for_status() return await resp.text() async def main(): urls = [f"https://example.com/detail/{i}" for i in range(1, 101)] async with aiohttp.ClientSession() as session: tasks = [fetch(session, url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) print(f"成功数: {sum(1 for r in results if isinstance(r, str))}") asyncio.run(main())

这段代码里有几个关键点需要解释：

• async with aiohttp.ClientSession() as session：可以类比成线程池版本里的 requests.Session。复用这个 session，可以重用底层连接池，而不是每个请求重新建立 TCP 连接，性能差距非常明显。
• session.get(url, timeout=10)：这里设置了每次请求的超时时间，防止某个服务器无响应时协程被永久挂起。这个坑我中过，不加 timeout 的并发协程，遇到一个黑点 IP 就能把整个任务卡住。
• asyncio.gather(*tasks, return_exceptions=True)：如果某个请求抛异常，默认会立刻中断整个 gather。加上 return_exceptions=True 之后，单个任务出错不会影响其他任务，错误会变成结果返回。这对于真实爬虫来说太重要了，因为网络环境里超时、拒连、反爬屏蔽都是常态。

3.3 加上并发限制：Semaphore 的用法

如果直接把 100 个任务全部塞给 gather，等于一瞬间发出 100 个 TCP 连接请求。本地可能没事，但对目标服务器压力很大，而且如果你用的是代理池、有限的文件描述符，分分钟崩掉。更科学的做法是用信号量控制同时进行的任务数：

import asyncio import aiohttp async def fetch(session, sem, url): async with sem: async with session.get(url, timeout=10) as resp: resp.raise_for_status() return await resp.text() async def main(): urls = [f"https://example.com/detail/{i}" for i in range(1, 101)] sem = asyncio.Semaphore(10) # 同一时刻最多 10 个请求 async with aiohttp.ClientSession() as session: tasks = [fetch(session, sem, url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) print(f"成功数: {sum(1 for r in results if isinstance(r, str))}") asyncio.run(main())

asyncio.Semaphore(10)做的事情就是维护一个计数器，每个协程进入async with sem:时计数器减一，退出时加一。当计数器归零，后续协程会在async with sem:这行被挂起，相当于排队等待前面某个请求完成。这个等待是在协程层面的，不会阻塞事件循环，所以其他正在等待 IO 的任务照样可以获得调度。

这里我给的并发数建议是 10，具体可以根据目标站点的承受能力调整。如果是对外开放的 API，可以适当放大到 50；如果抓的是小站点，5 都已经算很激进的了。

3.4 加入重试机制和流量控制

真实环境中，一次请求失败太常见了。我在协程版爬虫里通常会封装一个带重试的 fetch 函数。网上很多博客是直接递归调用协程，但我更推荐用循环实现：

async def fetch_with_retry(session, sem, url, retries=3): for attempt in range(retries): try: async with sem: async with session.get(url, timeout=10) as resp: if resp.status == 503: raise aiohttp.ServerDisconnectedError resp.raise_for_status() return await resp.text() except (aiohttp.ClientError, asyncio.TimeoutError) as exc: if attempt == retries - 1: print(f"URL {url} 最终失败: {exc}") raise wait_time = 0.5 * (attempt + 1) print(f"URL {url} 第 {attempt + 1} 次失败，{wait_time}s 后重试") await asyncio.sleep(wait_time) return ""

注意一点：await asyncio.sleep(wait_time)不会阻塞所有任务，当前协程只是让出了一段时间，其他协程还是会继续执行。这就是协程模式做重试比多线程模式更优雅的地方，重试等待本身不会占用额外线程资源。

3.5 数据落地：边爬边写还是最后统一写

很多人写爬虫的时候喜欢把所有结果攒在内存里，最后一次性存盘。如果结果不多，比如几千条，完全没问题。但如果数据量上了万，建议用队列加消费者协程的模式：一个协程专门负责产生任务，一个协程专门负责消费并把数据写入 CSV 或数据库，避免内存大爆炸。

对于本文这种入门级别的案例，最简单的方式是最后统一处理：

async def main(): results = [] ... results.extend(await asyncio.gather(*tasks, return_exceptions=True)) # 统一落盘 with open("output.txt", "w", encoding="utf-8") as f: for r in results: if isinstance(r, str): f.write(r + "\n")

如果你不想把所有内容都放内存，可以改用 asyncio.Queue：

queue = asyncio.Queue(maxsize=20) async def consumer(): while True: item = await queue.get() # 处理并写入文件 queue.task_done()

这种生产者-消费者模式的好处是爬取和落盘解耦，也是实际项目中比较成熟的结构。

4. 协程和 asyncio 的常见坑

网上的协程教程大多停留在"能跑就行"的阶段，但真正写进生产环境，一堆细节问题就会冒出来。我在不同项目里踩过不少坑，挑几个影响最大的说一说。

4.1 语法层面的隔离：不能和 requests 混着用

协程的美丽建立在事件循环保持畅通的基础上。如果在协程内部使用了requests.get()这种同步阻塞调用，整个事件循环会被卡住。因为阻塞调用是在当前线程执行，事件循环没法在等待期间去调度其他协程。这个问题的后果是极其隐蔽的：表面上代码没报错，但并发效果完全消失，所有请求串行执行，而且你从输出日志里很难察觉。

我有一次线上服务响应突然变慢，排查到最后，发现是某个协程里曾经混进了一条线程池代码，里面用了 requests。一条老鼠屎打乱一锅汤，整个事件循环的同事全在那等它。所以协程代码里，IO 操作必须用异步库：

• requests → aiohttp，或者 httpx 的 AsyncClient
• 标准库 socket 阻塞调用 → asyncio.open_connection
• 标准库 subprocess 阻塞调用 → asyncio.create_subprocess_shell

如果你确实需要在协程中调用同步阻塞函数，可以用asyncio.to_thread()把它丢到独立线程池，而不会阻塞事件循环：

result = await asyncio.to_thread(requests.get, url)

这种方式适合那种没有异步替代品的旧库，但要记住，这本质还是线程池，本质上是有线程调度开销的。

4.2 asyncio.run 每次都会创建新的事件循环

asyncio.run()是一个非常方便的入口，但它每次都会创建新的事件循环，任务执行完毕后会关闭它。这带来一个问题：如果你在一个事件循环里创建了某个连接池，比如 aiohttp.ClientSession，而这个 session 在当前事件循环结束后才被释放，就会出现 "Event loop is closed" 这样的异常。

实际开发中，我喜欢把整个会话的生命周期限制在同一个事件循环内，不要跨循环复用对象。像是这样：

async def main(): async with aiohttp.ClientSession() as session: # 在这个 session 生命周期内做完所有请求 ... asyncio.run(main())

不要在main()外面创建 session，再传进来。尽管有时候它碰巧能跑，但在不同的事件循环中复用底层 connect 对象，极易触发各种奇怪的 RuntimeError。

4.3 Task 和 Future 的区别

asyncio 里有两个容易混淆的概念：Task 和 Future。Future 是一个更底层的可等待对象，代表一个"未来会有结果"的操作；Task 是 Future 的子类，专门用来包装协程，并把它调度到事件循环中。asyncio.create_task()就是创建一个 Task，把协程立刻排入调度队列；而await coroutine则是直接把协程交给事件循环执行，直到完成，之间不再调度其他任务。

如果你希望多个协程并行执行，正确的做法是先用create_task()创建任务，再统一await：

task1 = asyncio.create_task(fetch(url1)) task2 = asyncio.create_task(fetch(url2)) await task1 await task2

如果你直接写：

await fetch(url1) await fetch(url2)

那 url1 跑完才会跑 url2，串行。这个坑特别容易踩，因为 Python 的语法看起来像是"await 了这个就去处理下一个"，其实不会。

也可以用asyncio.gather()替代上面创建多个 Task 再逐个 await 的写法，它会自动把协程包装成 Task 并发调度。两者区别在于，gather 能统一收集结果和异常处理，语义更接近"并发执行一批同类型任务"。

4.4 超时与取消：协程不是无脑挂着的

你不给协程设置超时，它可能真的永远挂着。某个服务端异常导致 TCP 连接不关闭，你的协程可能一直等响应，事件循环里所有并发协程都停在那里。很多初学者碰到这种情况的第一反应是"程序死了"，实际上是协程在等待一个永远不会到达的数据包。

解决方案是：

try: async with asyncio.timeout(5): result = await fetch() except TimeoutError: print("请求超时")

Python 3.11 引入了asyncio.timeout()，低版本可以用asyncio.wait_for()：

try: result = await asyncio.wait_for(fetch(), timeout=5) except asyncio.TimeoutError: print("请求超时")

注意，超时触发的瞬间协程会被取消，但它内部的 finally 块还是会执行。如果你的协程里打开了某些资源，别忘记用async with确保资源被正确释放，不然在大量超时场景下会出现资源泄漏。

4.5 不要滥用 return_exceptions=True

我前面推荐了asyncio.gather(*tasks, return_exceptions=True)，这是为了整个任务稳定运行。但要注意，加上这个参数后，所有的异常都变成了"结果"，你需要自己去甄别哪些是数据、哪些是异常。如果某个协程内部逻辑太复杂，异常被吞掉后反而会导致静默失败。

我的习惯是：在创建任务的入口处做一层异常捕获，这样 gather 里通常不会触发异常，return_exceptions 只是兜底：

async def safe_fetch(session, sem, url): try: return await fetch(session, sem, url) except Exception as exc: print(f"{url} 出错: {exc}") return None

这样 gather 的结果要么是数据，要么是 None，清理起来更容易。

5. 协程性能实测与适用边界

很多人以为协程万能，任何并发场景上协程都最优。这个想法不准确。协程在最合适的场景里确实能达到很高的资源利用率，但用错了场景，效果可能比同步还差。

5.1 我的 100 请求实测数据

自己在笔记本上做过一组对比，目标是同一个 API 端点，每次请求模拟 300ms 延迟（服务端 sleep），分别测同步、线程池、协程三种模式：

耗时上协程和线程池几乎拉不开差距，因为瓶颈都在服务器响应延迟。协程的真正优势在内存占用。当并发达到 500 时，线程池内存占了 800MB 以上，协程组只有 150MB 左右，差距非常明显。另一个实测里，网络延迟越大，协程对线程池的优势越明显，因为协程的等待成本几乎为零，而线程池还需要维持大量线程。

5.2 协程不适合的场景

有几类场景协程没啥优势。第一是 CPU 密集型任务，比如图像处理、复杂数值计算、视频编码，这种任务需要的是真正的多核并行，协程单线程没法加速，用了甚至有反效果。应该用 multiprocessing 或者直接上 PyPy、Numba 这类工具。

第二是单次 IO 操作耗时极短且延迟极低的场景。比如内存数据库缓存操作，本来就微秒级完成，引入事件循环的开销可能比实际 IO 时间还长，得不偿失。

第三是对第三方库不熟悉的场景。如果目标库只提供同步接口，你又不想把代码拆得很复杂，那用线程池可能更省事。协程要求整个调用链都是异步的，这通常意味着对生态的改动比较大。

5.3 网络编程中协程库选型建议

做网络 IO 时，最常用的还是 aiohttp，它同时支持客户端和服务端，API 稳定，资料多。使用过程中有一点值得注意的是，aiohttp 在底层使用连接池，每个域名默认会复用连接，因此并发请求同一个站点时，实际建立的 TCP 连接数远比任务数少，通常更不容易触发站点端的连接限制。

如果只是做 HTTP 客户端，还可以看 httpx，它同时支持同步和异步接口，代码迁移成本很低。在协程模式下，httpx.AsyncClient 用起来和 requests 几乎一样舒适：

import httpx async def fetch_httpx(client, url): resp = await client.get(url) resp.raise_for_status() return resp.text

如果你要处理非常底层的 TCP/UDP，asyncio 自带的 open_connection、start_server 反而比任何第三方库都可靠。协程的优势并不仅仅体现在 HTTP 场景，任何涉及网络 socket、Redis、数据库连接的 IO 等待场景都适用。

5.4 什么时候用 gather，什么时候用 Queue

我比较喜欢的一个选择规则是这样的。如果是"一批独立任务并发执行，等待全部完成"，用 gather 最简单，代码最少，逻辑最清晰。如果任务是"持续不断地来，每个任务处理耗时不定，需要控制背压"这种流式场景，用 asyncio.Queue 更合适。如果你明确希望某个任务先完成先处理，后完成无所谓顺序，比如先到的请求优先处理，用 asyncio.as_completed 可以把完成顺序排序，而不是等最慢的那个。

for coro in asyncio.as_completed(tasks): result = await coro # 先完成的先处理

这套组合用顺手之后，基本能覆盖绝大多数网络并发编程需求。

6. 我实际调试协程代码的方式

协程代码的调试和同步代码不太一样。同步代码出 bug，断点一打就能看到当前调用栈。协程代码里，报错往往发生在非常"奇怪"的时机，比如"在事件循环结束后调用某个方法"、"Task got Future attached to a different loop"，这类信息看起来云里雾里。我总结了一套自己的调试方法，对于刚接触协程的人来说比较友好。

6.1 打印日志 + 标识当前协程

在开发阶段，我习惯在每个协程的第一行加一个带有任务标识的 print。谁开始运行、谁在等待、谁结束，一目了然：

async def fetch(session, sem, idx, url): print(f"[{idx}] 开始") async with sem: print(f"[{idx}] 拿到信号量") async with session.get(url, timeout=10) as resp: data = await resp.text() print(f"[{idx}] 结束") return data

通过这段日志，可以直观看到并发是否生效：如果所有"开始"都一股脑出现在"拿到信号量"之前，信号量数量是对的；如果严格串行，那就说明代码里哪里有阻塞调用混进来了。

6.2 开启 asyncio 调试模式

Python 官方提供了调试模式，可以检测到未等待的协程、调度的延迟等。用环境变量或者代码开启：

asyncio.run(main(), debug=True)

或者设置环境变量PYTHONASYNCIODEBUG=1。在这个模式下，当你忘记 await 一个协程，事件循环会输出警告。这个提醒对我这种经常犯低级错误的人来说是救命稻草。

6.3 用 asyncio.Event 控制暂停和恢复

写多阶段任务的时候，比如一个爬虫先登录拿到 token，再带着 token 去抓取，中途需要等用户确认某个条件，asyncio.Event()非常好用。它比 threading.Event 好用太多，因为等待它是完全非阻塞的：

ready = asyncio.Event() async def wait_flag(): print("等待标志...") await ready.wait() print("标志已设置") async def set_flag(): await asyncio.sleep(2) ready.set()

这种方式能写出非常清晰的状态机逻辑，比用 while+await sleep 轮询优雅太多。

6.4 嵌套事件循环的坑

在 Jupyter Notebook 里跑 asyncio 是最常见的嵌套事件循环坑。Notebook 本身已经跑着一个事件循环，你再调用asyncio.run()就会报 "This event loop is already running"。处理方式有两种：用nest_asyncio补丁，或者把逻辑全部塞到一个 main 协程，用await main()调用而不是 asyncio.run。我推荐后者，写代码时就更接近生产环境的结构。

7. 协程之外：与线程、进程配合的正确关系

协程不等于完全替代线程和进程，而是用在不同层次上。在一个多进程架构中，每个进程都可以跑自己的事件循环，进程内部再用协程处理大量 IO 并发任务。对于极端场景，比如既要跑满所有 CPU 核，又要同时处理高并发网络请求，这种混合架构才是标准解法：multiprocessing 负责水平扩展，asyncio 负责单进程内的 IO 密集调度，再配合少量线程处理没有异步替代品的阻塞调用。

7.1 信号处理与优雅退出

生产环境里协程服务需要有优雅退出能力。收到 SIGINT 信号时，应该停止接收新任务，等待当前任务完成后再退出。asyncio 里可以通过捕获信号然后设置事件循环的某个 Event 来实现：

loop = asyncio.get_running_loop() stop_event = asyncio.Event() def on_sigint(): print("收到退出信号，正在优雅关闭...") stop_event.set() loop.add_signal_handler(signal.SIGINT, on_sigint) await stop_event.wait()

这套逻辑配合 asyncio.run(main()) 的上下文管理，能让代码在 Ctrl+C 时不会留下一片乱七八糟的半截数据。

7.2 监控与指标采集

协程并发程序有一个特点：活跃协程数是个很核心的健康指标。如果某个环节阻塞，活跃协程数会快速上升。我一般会在循环里周期性打印或者上报这个数值作为监控：

all_tasks = asyncio.all_tasks() running = [t for t in all_tasks if not t.done()]

不过要注意，在不同事件循环里调用 asyncio.all_tasks() 只能看到当前事件循环的任务，所以这个监控代码最好放在被监控的协程内部。

7.3 未来的方向：结构化并发

Python 协程发展到现在，虽然功能已经很完善，但有一点始终不如 Go 的 goroutine 优雅：没有内置的"协程组"概念，管理大量生命周期相互关联的协程很麻烦。社区里有人开始推 trio、anyio 这类结构化并发库，anyio 在 asyncio 之上提供更严谨的任务取消和生命周期语义。如果你要在协程项目里做复杂的多任务编排，可以了解一下 anyio 的 task group 模式，写出来的代码更清晰，也更好维护。

我自己最近的新项目里已经切换到了 anyio 做高层封装，底层仍然是 asyncio 在跑。它兼容 asyncio.run 和 trio.run 两套后端，任务取消做到"要么全部取消，要么全部完成"，这种确定性对工程来说太重要了。

收尾：我的一点实际体会

做了这些年网络相关的开发，我越来越觉得协程不是什么高深概念，它不过是对"人类做事情的直觉"的一种还原。你在等外卖的时候不会死盯着页面，你在等接口返回的时候也不再需要把整条线程绑死。写协程代码真正要克服的不是语法，而是思维习惯：从"命令式地一步步执行"切换成"用事件循环去编排多个步骤之间的等待关系"。

如果你刚开始学，我建议不要一开始就花大量时间背 API，先把事件循环、await 挂起、Task 调度这三件事彻底搞懂。把一个简单的并发爬虫跑透，把每个异常都弄明白，比看十篇教程都管用。等你能自如地用信号量控制并发、用超时兜底、用事件协调任务状态时，你会发现在 Python 里写网络并发程序，完全可以写得比很多多线程代码更稳、更省资源，也更让人愉快。