企业官网建设流程全解析

很多人刚开始学 Python 时，都会在某个阶段被asyncio这套异步 I/O 库搞得晕头转向。网上能查到大量教程，但大多数要么讲得过于底层，上来就扔给你事件循环、协程调度、Future 对象这些概念；要么就是只给三个例子，你照着抄完还是不知道什么时候该用、怎么排查问题。作为一个在爬虫、消息队列、API 网关服务里实际踩过不少坑的 Python 使用者，我打算用这篇博文把 asyncio 从“是什么、为什么”讲到“怎么写、怎么踩坑”，尽量让一个刚入门的读者也能真正理解并跑通自己的异步代码。

先给一个结论：asyncio是 Python 提供的异步 I/O 编程框架，核心是帮你在单线程内处理大量并发的 I/O 操作，比如网络请求、文件读写、数据库查询、子进程调用。它特别适合 I/O 密集型任务，对 CPU 密集型任务基本没有帮助。如果你是写爬虫、写 API 服务、写消息消费者、写需要同时等一堆外部服务响应的脚本，那 asyncio 几乎是你绕不开的工具。这个内容适合所有已经能写简单 Python 脚本、但还没搞懂async/await和事件循环的开发者，或者已经在用线程池但总感觉并发上不去、资源浪费严重的人。读完之后，你不仅会写异步代码，还会明白那些写法背后的调度逻辑。

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

想理解 asyncio，先得理解阻塞。你的程序每一次发 HTTP 请求、读本地文件、连数据库、调用第三方 API，本质上都是在等“别人”。等的过程里 CPU 其实什么事都没干，但写同步代码时，整个线程就被卡住了，后面的代码只能排队。

同步代码像是只有一个收银员的超市，顾客 A 买了十分钟东西，收银员就站那儿等十分钟，后面所有人陪着等。线程池方案像是多开了几个收银员，每个收银员独立服务一个顾客，看起来并行处理了，但每个收银员依然会花大量时间干等顾客掏钱。asyncio 的方案则是一个收银员同时服务一堆顾客：A 说“我要买大米”，收银员把 A 的购物车清单登记好，转身去帮 B 结算；等 A 把米搬过来了，收银员再回来收钱。收银员全程没有闲着，CPU 也是这样。

1.1 I/O 密集型 vs CPU 密集型任务

asyncio 不解决所有并发问题。它只解决 I/O 密集型的场景。

I/O 密集型任务的特点是：程序大部分时间在等待外部资源，CPU 计算量非常少。比如网络爬虫等待服务器返回 HTML、脚本等待数据库 exec 返回结果、大量调用 REST API。对这类任务，异步单线程往往比多线程更快，因为线程切换和锁的开销都没有了。

CPU 密集型任务是另一回事，比如视频编码、大矩阵计算、复杂的加密解密、数据清洗里的循环重计算。这类任务真正吃的是 CPU 算力，如果你把一个 CPU 密集任务丢进 async 函数里，它反而会阻塞事件循环，导致其他协程全部无法执行。这时候正确选择是多进程（multiprocessing）或者把计算密集部分交给run_in_executor里的进程池，而不是 asyncio 本身。

区分方法很简单：如果任务耗时主要花在await一个外部调用上，它是 I/O 密集，适合 asyncio；如果耗时主要花在纯 Python 计算循环里，它是 CPU 密集，不适合 asyncio。我见过不少人拿 asyncio 去跑分页解析 Excel 或者做大量正则替换，结果发现比同步版本还慢，就是因为走了错误的赛道。

1.2 同步阻塞带来的真实痛苦

举一个很常见的场景。你需要从 100 个不同的 API 拉取数据，然后汇总入库。同步代码的写法是：

import requests def fetch_sync(): results = [] for i in range(100): # 每个请求耗时约 1 秒，这里会依次等待 resp = requests.get(f"https://api.example.com/item/{i}") results.append(resp.json()) return results

如果每个请求耗时 1 秒，100 个请求就是 100 秒。这 100 秒里，程序绝大部分时间都在傻等网络响应，CPU 占用率几乎为 0。换用多线程会好转，但 100 个线程带来的上下文切换、GIL 限制、线程安全问题又会出现。

import asyncio import httpx async def fetch_async(): async with httpx.AsyncClient() as client: tasks = [] for i in range(100): tasks.append(client.get(f"https://api.example.com/item/{i}")) responses = await asyncio.gather(*tasks) return [resp.json() for resp in responses]

这段代码发 100 个请求时，不是一个个等，而是同时发出 100 个请求，在不违反目标服务器限额的前提下，总耗时能压缩到大约 1 秒多，也就是“最慢那一个请求”的耗时。这就是 asyncio 的价值。

2. async/await 核心机制与事件循环

从语法上看，asyncio 代码就多了async def和await两个关键字。但很多人第一次写就摔跟头，就是因为没搞懂背后那套调度机制。下面把核心机制一块一块拆开讲。

2.1 协程不是线程，也不是普通的“语法糖”

协程（coroutine）是一个可以暂停和恢复的函数。当你用async def定义一个函数时，调用它并不会真正执行函数体，而是返回一个协程对象。这个对象在被 await 或加入事件循环之前，函数体一行代码都不会运行。

async def say_hello(): print("hello") return 1 # 这样并不会打印 hello coro = say_hello() print(type(coro)) # <class 'coroutine'> # 必须 await 或加入事件循环

这跟普通函数完全不同，也是新手最常见的困惑来源。普通函数调用即执行，而协程调用只是“把菜谱写好”，真正做饭要等事件循环来叫它。

线程是操作系统管理的，每条线程有自己的栈和寄存器状态，切换由内核调度，切换成本高。协程则完全在用户态运行，切换只发生在遇到await时。因为切换不涉及系统调用，协程之间切换的开销可以低到微秒级，所以单线程里同时跑数千个协程完全没问题。

2.2 事件循环的调度逻辑：await 什么时候让出

事件循环是整个 asyncio 的发动机。它是一个无限循环，不断做三件事：拿到当前可执行的协程、执行它、看它是否让出执行权。当协程遇到await某个可能耗时的操作时，它会向事件循环注册一个“我还在等，等到了叫我”，然后立即跳回事件循环，让其他协程执行。

import asyncio async def worker(name, delay): print(f"{name} 开始") await asyncio.sleep(delay) print(f"{name} 结束") async def main(): await asyncio.gather( worker("task-a", 2), worker("task-b", 1), ) asyncio.run(main())

运行一下你会发现顺序是：task-a 开始、task-b 开始、task-b 结束、task-a 结束。task-a先执行到await asyncio.sleep(2)时暂停，事件循环接管，让task-b开始执行。sleep 本身不是真的“卡住”，而是告诉事件循环，2 秒之后再来叫我。等待期间事件循环干了很多事。

这个行为也可以解释为什么 asyncio 里不允许写太重的同步逻辑。如果你在某个协程里写了一个time.sleep(2)或者一段会运行 5 秒的计算循环，事件循环是没法“抢走”执行权的，因为事件循环本身也是在同一线程里。它必须等这个协程自己主动让出。只有当代码执行到await时，才会切回事件循环。

2.3 tasks、coroutines、futures 三者关系

三个概念常被混用，但理解它们能省很多调试时间：

• Coroutine（协程对象）：async def函数调用后得到的对象，本身还不能独立调度，必须等 await。
• Task（任务）：包装协程并把它提交给事件循环调度的对象。一个 Task 对应一个被调度的协程。
• Future：一个更低层的对象，表示一种“未来可能完成的结果”。Task 是 Future 的子类，但日常开发中你不一定直接碰 Future，大多数时候你处理的是 Task。

有一段代码很直观：

import asyncio async def simple(): return "done" async def main(): task = asyncio.create_task(simple()) print(task) # <Task pending ...> result = await task print(result)

asyncio.create_task(coro)是显式地把协程注册到事件循环中。之后即便你不立即 await，这个协程也已经开始调度执行了。注意，协程必须被调度才会运行，光创建一个协程对象没有任何并发效果。如果你在main()中连续创建了两百个 task 然后各自await，它们就会交替执行。

3. 日常开发最常用的 asyncio API

asyncio 的 API 数量不少，但日常开发高频用到的其实就那几个。把它们彻底吃透，比背一百个 API 更管用。

3.1 gather、wait 和 as_completed 的区别

并发执行多个协程时，最常用的方法是asyncio.gather。它接收一批协程对象或 Task，返回一个列表，列表顺序与传入顺序一致，即使某个协程后完成，它在结果列表里也还是在对应位置。

import asyncio async def fetch_url(name): await asyncio.sleep(1) return name.upper() async def main(): results = await asyncio.gather( fetch_url("aaa"), fetch_url("bbb"), fetch_url("ccc"), ) print(results) # ['AAA', 'BBB', 'CCC']

gather有一个容易被忽略的行为：如果不传return_exceptions=False，一旦某个协程抛异常，它会立即把异常抛给 await 的一方，同时其他协程并不会自动停止，但 gather 的结果可能不会再返回。如果你不想因为一个失败导致整个 gather 被中断，可以传return_exceptions=True，异常会作为结果元素返回，方便逐个判断。

asyncio.wait和 gather 不一样，它更强调等待集合，并且支持FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETED这些模式。比如你要“等这批任务里最早完成的那个”，wait 更合适。gather 的返回结果不包含任务本身状态，而 wait 返回(done, pending)两组任务集合，自由度更高。

asyncio.as_completed则是流式处理。它接收一批任务，返回一个迭代器，每完成一个任务就会 yield 一个结果，所以你可以边等边处理，而不是全跑完再统一拿结果。适合类似“100 个请求陆续完成，每完成一个就立即解析和写入数据库”的场景。

3.2 run_in_executor：把同步代码变成异步

asyncio 只能让自己的协程实现异步，但它无法把任意一个普通同步函数变成非阻塞。比如你用requests.get()这种同步库发请求，直接放在协程里就是阻塞的。

解决办法是用loop.run_in_executor，它能将同步函数放到线程池或进程池里执行，从而不阻塞事件循环。

import asyncio import requests def sync_request(url): return requests.get(url).status_code async def async_context(): loop = asyncio.get_running_loop() # 使用默认线程池执行同步请求 result = await loop.run_in_executor(None, sync_request, "https://api.example.com/health") print(result)

在 Python 3.9 之后的代码里，官方推荐用asyncio.to_thread(func, *args)达到同样的效果，它的写法更简洁。

import asyncio import requests async def main(): result = await asyncio.to_thread(requests.get, "https://api.example.com/health") return result.status_code

这里要注意，run_in_executor可以传ThreadPoolExecutor或ProcessPoolExecutor实例。如果任务是 CPU 密集型，请用ProcessPoolExecutor，因为多线程在 GIL 下没法利用多核；如果任务主要是 I/O 密集，默认的线程池就够了。

3.3 asyncio.Queue：控制并发、传递数据

当任务数量特别大时，一次性创建几百个 Task 可行，但创建几万、几十万个就有点危险了，事件循环维护它们也会吃力。更合理的做法是用生产者消费者模式，搭配asyncio.Queue控制并发和传递数据。

队列一个有界对象，可以设定maxsize。生产者往队列里put数据，消费者从队列里get数据。队列的put和get都是异步方法，在队列满时会自动让出执行权，队列空时消费者也会等待，而不是空转。

import asyncio import random async def producer(queue, total): for i in range(total): await queue.put(i) print(f"生产: {i}") await asyncio.sleep(0.2) # 发送终止信号 await queue.put(None) async def consumer(queue, name): while True: item = await queue.get() if item is None: await queue.put(None) # 继续传递终止信号给下一个消费者 break print(f"消费者 {name} 处理: {item}") await asyncio.sleep(random.uniform(0.1, 0.3)) async def main(): q = asyncio.Queue(maxsize=10) producers = [asyncio.create_task(producer(q, 20))] consumers = [asyncio.create_task(consumer(q, f"c{i}")) for i in range(3)] await asyncio.gather(*producers) await asyncio.gather(*consumers) asyncio.run(main())

这里有个细节：多个消费者同时从一个队列取数据时，每个消息只会被消费一次，asyncio.Queue 天然是线程安全的协程安全容器，不需要额外加锁。队列的join()方法可以等待队列清空，是生产者消费者模式中常用的同步手段，但与简单 get 不同，你需要同时task_done()。

3.4 超时管理与取消协程

真实场景里，网络请求可能会卡住很久。如果不设超时，协程可能一直挂着。大多数人会用asyncio.wait_for：

import asyncio async def slow_work(): await asyncio.sleep(100) return "done" async def main(): try: result = await asyncio.wait_for(slow_work(), timeout=3) except asyncio.TimeoutError: print("超时了")

注意这里的行为：超时发生后，被等待的协程内部会被取消，也就是CancelledError会在await处被注入。如果协程捕获并吞掉了 CancelledError，wait_for是没法真正终止它的，任务会继续跑。这种情况被称为“任务泄漏”，是异步服务里比较隐蔽的资源浪费。

正确清理自己的协程需要用到try/finally。如果协程持有数据库连接或文件句柄，在 finally 里释放是必须的操作。下面展示一个协作式取消的写法：

async def worker(): try: while True: await asyncio.sleep(1) print("工作中") except asyncio.CancelledError: print("收到取消通知") raise # 一定要重新抛出，才算真正取消

细节在于，except asyncio.CancelledError里完成资源清理后要重新raise，否则事件循环会认为任务已被取消但协程没有正确响应，可能导致Task was destroyed but it is pending之类的警告。

4. 实战：异步请求调度怎么写才能稳定

理论讲完，来做一个相对完整的实战。目标是用 asyncio 实现一个“可控并发、有重试、有超时”的请求调度器。这个模式在爬虫、批量查询、报表回填任务里都能复用。

4.1 选择异步 HTTP 客户端

Python 里主流的异步 HTTP 客户端有两个：aiohttp和httpx。 aiohttp 在 asyncio 生态里出生得更早，功能完善，支持 WebSocket、连接池，但 API 风格略老。 httpx 提供同步和异步两套接口，如果你已有requests使用经验，上手 httpx 会平滑很多。我个人在写新项目时更倾向用 httpx，因为它对 HTTP/2、代理、超时配置更友好，而且和 requests API 相似，不太容易踩惯性思维的坑。

安装很简单：

pip install httpx

4.2 一个可控并发的请求调度器

先定义一个异步 fetch 函数，包含超时和简单重试。我建议把“单次请求”封装成一个函数，后面加并发、加信号量都容易扩展。

import asyncio import httpx async def fetch_with_retry(client, url, retries=3, timeout=5.0): for attempt in range(retries): try: resp = await client.get(url, timeout=timeout) if resp.status_code == 200: return resp.text # 4xx 大多数不用重试，5xx 和网络错误才需要重试 elif resp.status_code < 500: return None except (httpx.TimeoutException, httpx.NetworkError) as e: if attempt == retries - 1: print(f"{url} 请求失败: {e}") return None await asyncio.sleep(0.5 * (attempt + 1)) # 简单的线性退避 return None

接着用asyncio.Semaphore控制并发数。因为一次性发起几千个并发请求很容易把目标服务打爆，也可能被自己的操作系统文件描述符限制挡住。

async def bounded_fetch_all(urls, max_concurrency=10): semaphore = asyncio.Semaphore(max_concurrency) async with httpx.AsyncClient() as client: async def one_task(url): async with semaphore: return url, await fetch_with_retry(client, url) results = [] for task in asyncio.as_completed([one_task(url) for url in urls]): url, content = await task results.append((url, content)) if content is None: print(f"记录失败: {url}") return results async def main(): urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(50)] results = await bounded_fetch_all(urls, max_concurrency=8) print(f"完成: {len(results)}")

这里asyncio.Semaphore的本质是限制同一时刻能进入临界区的协程数。当并发任务多时，信号量内部会让超出的协程等待。把信号量的获取写在每个任务的入口处，可以保证并发不会超过设定值。

4.3 如何优雅处理失败和中间结果

在实际任务里，我不建议把所有结果都攒到内存里再处理，数据量大时很容易 OOM。使用asyncio.as_completed或queue边取边存更合理。上面的例子用了 as_completed，一旦某个请求完成，立即可以把结果写到文件、数据库或传给下一个处理函数。

如果你需要严格保持提交顺序，比如批量更新第 i 条记录的响应要写回第 i 条的状态，那么gather的一一对应结果列表更方便。两者没有绝对优劣，按数据消费方式选择。

还要注意，异步 HTTP 客户端默认复用连接池。如果你在循环里反复创建AsyncClient，连接不会被复用，每次握手都是新的 TCP 连接，性能会差很多。应该在一个会话生命周期内复用一个 client，这也是上面代码把一个 client 作为参数传给 fetch 函数的原因。

5. 异步调试与常见坑速查

asyncio 项目跑起来之后，经常出现“代码看起来没问题，但就是卡死或疯狂报错”的情况。下面整理一些我实际踩过、也帮同事排查过的常见问题，表格和注释都给了排查思路。

5.1 在同步函数里调用 async 函数时报错

很多人在 Flask/Django 的视图函数里直接async_func()，发现没有任何输出，或者干脆报RuntimeWarning: coroutine was never awaited。原因再强调一次：调用 async 函数返回的是协程对象，不会执行。

在同步函数里想运行一个 async 函数，最简单的做法是：

import asyncio def sync_func(): asyncio.run(some_async())

但注意，在已经有一个运行中事件循环的线程里再次调用asyncio.run()会报RuntimeError: asyncio.run() cannot be called from a running event loop。这时不要硬调，而是把异步逻辑交给asyncio.run_coroutine_threadsafe，或者在你的异步框架里单独开线程使用独立事件循环。

5.2 有一个协程卡住，其他全部不动

事件循环被阻塞了十有八九是某个协程里使用了同步阻塞调用，比如time.sleep()、requests.get()或者大计算循环。很多人打印日志时发现第一个任务跑完，第二个任务迟迟不开始，就以为是并发没生效，实际是事件循环被第一个任务的同步操作卡死了。

排查思路：在任务日志里每隔一段时间打印当前所有任务状态，用asyncio.all_tasks()可以看到事件循环里还存在哪些任务。

import asyncio async def monitor(): while True: tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] for t in tasks: print(t, t._state) await asyncio.sleep(5)

不过_state是私有属性，不建议在产品代码里用。更稳妥的方式是把关键业务流程包一层带超时的asyncio.wait_for，超时后直接打日志。

5.3 事件循环选型：run 还是 get_event_loop

Python 3.10 之后，官方对事件循环的管理做了清理。以前常见的loop = asyncio.get_event_loop(); loop.run_until_complete(main())这种写法，在新版本里容易碰到DeprecationWarning。 对绝大多数应用场景，直接使用：

asyncio.run(main())

asyncio.run()会自己创建事件循环、运行 main 协程，最后关闭事件循环，省去手动管理资源的问题。它可以多次调用，每次调用都会新建一个循环。但要注意，它不可以在一个已经运行的事件循环里嵌套调用。如果是在 Jupyter Notebook、既有 event loop 的服务里，更推荐await main()或通过nest_asyncio打补丁（非必要不用，补丁会比较粗暴）。

5.4 多线程里同时跑 asyncio 的问题

asyncio 是单线程的事件循环，但有时你确实需要在一个多线程程序的子线程里各自跑一个异步任务。此时核心原则是：同一时刻一个线程只允许有一个事件循环，不同线程的事件循环不能互相交叉操作 Task。 错误示范是把线程 A 里创建的 Task 传给线程 B 等待，这会导致不稳定。正确做法是让线程 A 只负责运行自己的事件循环，线程间数据用queue.Queue或concurrent.futures传递。

5.5 调试异步代码的工具建议

没有哪个人能只看日志就解决所有异步问题。我最常用的工具组合如下：

• asyncio.run(main(), debug=True)：开启调试模式，会显示任务执行耗时、未 await 的协程、回调执行时长；
• PYTHONASYNCIODEBUG=1环境变量：效果同上；
• faulthandler：程序卡住时发送SIGABRT打印线程栈，能定位到底是哪一行没让出事件循环；
• pytest-asyncio：写单元测试时模拟异步逻辑的标准方案，比自己在测试里手写asyncio.run更优雅。

5.6 常用异常情况速查

6. 扩展思考：asyncio 与线程池搭配的正确姿势

asyncio 不是银弹，它在实际项目里经常和线程、进程池一起使用，组合得当能发挥最大的威力。比如一段业务逻辑里，既要调外部 HTTP API，又要把返回结果交给本地一个 CPU 密集的解析函数，最合理的结构是：HTTP 部分用 asyncio，CPU 密集部分用asyncio.to_thread或run_in_executor(ProcessPoolExecutor)执行，避免阻塞事件循环。通过这种方式，资源利用率通常会比单纯用线程池高不少，也比“同步代码 + 大量线程”的并发模型更可控。

我之前做过一个 RPA 任务调度器，需要同时监控几百个业务流程的状态，每个流程都要周期性地查数据库、调接口、等固定时间。如果给每个流程开一条线程，几百条线程既是系统负担，也让状态同步变得异常困难。后来把所有流程改成协程，定时等待用asyncio.sleep，数据库查询的同步客户端包进to_thread，整体代码量更少、并发能力也上去了。对我个人来说，这种“异步驱动、同步适配”的混合模式，才是 asyncio 最值得使用的姿势。

最后再分享一个实操里的小技巧：新手可以从async/await + httpx + asyncio.gather这个最小组合开始练手，先写一个能并发抓取 20 个网页的程序，再把队列、信号量、重试一个个加进去。遇到卡住就开debug=True跑一遍，绝大多数问题都会自己暴露出来。踩过几次坑之后，你再看 asyncio 就很容易形成直觉了。