很多刚接触Python的朋友，一听说“异步编程”就觉得高深莫测，再看到asyncio官网文档里那一大堆概念，直接劝退。但我在实际项目里用过几年asyncio之后，反而觉得它比多线程要好理解得多——只要抓住了“事件循环”和“协程”这两条主线，剩下的都是围绕它们展开的工具。这篇文章就是我从入门到能独立写出并发爬虫、异步API调度脚本的完整经验总结，专治各种“看文档看不懂、一跑就报错、并发一高就慢”的毛病。无论你是刚学完Python基础语法，还是在写一些IO密集型的工具脚本，都能从里面直接拿到能跑、能用的套路。
我先说说上手asyncio之后最明显的变化：以前写爬虫，几百个请求用requests一个一个等，一个半小时都跑不完；换成aiohttp加asyncio之后，十几分钟就全部搞定，而且代码量没有增加多少。这种“肉眼可见的提速”就是我写这篇东西的初衷。下面直接进入正题，我把从零理解asyncio到落地使用的全过程拆成几大块，每块都附了可以直接复制的代码和分析思路。
1. 理解异步编程之前，先搞懂它要解决的问题
1.1 同步代码的最大瓶颈：等待时CPU在空转
咱们先看一个最典型的同步场景：从网络上下载5个文件，每个文件下载需要2秒。用同步代码写，就是依次下载，总耗时10秒。这个过程中，程序在每次requests.get()之后，就傻傻地等待服务器的响应，这个等待期间，CPU其实什么也没干，就是线程挂起在那。如果这5个请求发生在真实生产环境里，那服务器返回的快慢、网络波动、DNS解析时间全都会串在一起，体验就是“慢”。
我常给朋友打个比方：同步代码就像在银行柜台办业务，只有一个窗口，前面的人没办完，后面的人就得一直排着。哪怕后面的人只想查个余额，也得干等。这里面真正耗时的是“等待”而不是“计算”，但同步模型没法把等待时间利用起来。
1.2 多线程解决了并发，却带来了新麻烦
很多人第一反应是：那我用多线程不就行了？确实，Python里可以用threading或ThreadPoolExecutor把等待时间并行出去。但问题在于：
- Python的
GIL（全局解释器锁）对CPU密集型任务并不友好，多线程并不能真正利用多核。 - 线程切换有开销，线程多了会疯狂占用内存。
- 线程之间的共享数据要加锁，一旦加锁，复杂度就上来了，容易出死锁、竞态这种“幽灵Bug”。
相比之下，asyncio走的是另一条路：它不需要操作系统切换线程，而是在一个线程内部，通过事件循环不停地在IO等待和代码执行之间切换。这就像把一个窗口的工作人员变成了一个超级调度员：只要有客户在填单子、等银行后台审核，调度员就立刻去服务下一个已经准备好的人。整个银行只有一个窗口，但没人闲着。
1.3 异步编程最适合什么样的任务
asyncio不是万能药。它特别适合IO密集型任务，也就是那些大部分时间在等网络响应、读文件、查数据库的操作。典型的有：
- 网络爬虫并发抓取
- 大量HTTP请求的批量调用
- 消息队列消费、WebSocket客户端
- 异步读写文件
反过来，如果你的任务全是CPU计算，比如视频转码、图像处理、大数计算，那asyncio帮不上什么大忙，甚至可能因为切换开销变得更慢。这种场景就应该考虑多进程（multiprocessing）配合核数去跑。
asyncio的核心价值，就是在单线程内把“等待”和“计算”交错进行，让程序在同样的时间内做更多事。明白了这一点，接下来学的每个概念都会变得顺理成章。
2. asyncio的四个核心概念，一步步拆给你看
2.1 事件循环：程序的调度中枢
事件循环（Event Loop）是asyncio的心脏。你可以把它理解成一个不断循环的“任务调度器”。它维护着一个任务队列，按顺序检查每个任务的状态：如果任务在等待IO，就把它挂起记录一下，先去执行其他可以执行的任务；如果任务的IO完成了，马上把它恢复继续跑。
在Python 3.7之前，你得手动用asyncio.get_event_loop()之类的方法去操作事件循环。Python 3.10以后官方推荐直接使用asyncio.run()，这个方法内部会帮你创建事件循环、运行协程、关闭事件循环，一步到位。所以新手根本不需要直接碰底层事件循环，只需要记住：我们写的协程，最终是扔给事件循环去调度的。
这里有个很多入门教程不会细说但很关键的点：事件循环默认运行在主线程中。如果你尝试在事件循环运行期间，再去运行另一个事件循环，就会报“This event loop is already running”的错误。这也是很多新手在用了asyncio.run()之后，又想在其他线程里启动异步任务时最容易踩的坑。
2.2 协程：可以暂停再恢复的函数
协程（Coroutine）是异步编程的基础单元。用async def定义的函数，在调用时不会立即执行函数体，而是返回一个协程对象。只有当你把它放到事件循环里运行，或者用await关键字等待它时，函数体里的代码才会真正开始跑。
看这个例子：
async def hello(): print("你好") await asyncio.sleep(1) print("一秒后继续") # 直接调用不会输出任何内容 coro = hello() print(type(coro)) # <class 'coroutine'>注意，这里hello()返回的是一个协程对象，你如果直接丢弃这个对象，程序会输出一个RuntimeWarning，提醒你协程没有被等待。正确的做法是用await去等待它，或者用任务把它包装起来。
await的作用就是告诉事件循环：我在这里要暂停，去等待某个耗时操作的结果。在await期间，事件循环就可以切换去执行别的协程。这就是异步代码能够“并发”的根本原因。
2.3 Task（任务）：让协程真正被调度起来
协程对象本身只是一个“空壳”，把它封装成Task之后，事件循环才会把它纳入调度系统。Task可以在后台运行，可以被取消，可以查询状态。创建Task最常用的方式有两种：
task = asyncio.create_task(hello()) # 创建但未马上运行 await task # 等待任务完成第二种是用asyncio.gather()同时创建并等待多个协程：
results = await asyncio.gather( hello("A"), hello("B"), hello("C") )gather()是入门阶段最推荐使用的并发运行函数。它会把传入的多个协程/任务包成一个新的协程，并且并发执行。等所有任务都完成后，会返回结果列表，顺序和传入顺序一致。这个“返回顺序一致”的特性在实际用的时候特别方便，你不需要自己去管理结果和任务的对应关系。
2.4 Future：未来时的结果占位符
Future这个概念在入门时可以稍微放一放，但为了后面排查问题，还是要了解一点。Future代表一个未来的、还没有计算完成的结果。Task就是Future的一个子类。其实你绝大多数时候都在跟Task打交道，不需要手动操作Future，但在实现底层协议、配合第三方库时，你会看到很多Future出现。只需要记住：一个协程被执行时，包装成的Task就是一个Future，这个Future会绑定协程运行的结果，一旦完成，事件循环会立即恢复等待它的那个协程。
这四个概念串起来，用一句话概括：async def定义协程，await让出控制权，asyncio.run启动事件循环，事件循环把协程包装成Task并调度运行。
3. 第一个异步程序：从代码运行看背后发生了什么
3.1 同步版示例，感受“傻等”是啥感觉
先写一个模拟IO操作的同步程序：
import time def fetch_data(url): print(f"开始请求 {url}") time.sleep(2) # 模拟网络等待 print(f"结束请求 {url}") return f"数据 {url}" start = time.time() for url in ["A.com", "B.com", "C.com", "D.com", "E.com"]: fetch_data(url) print(f"耗时: {time.time() - start:.2f}秒")运行结果大概是耗时10秒。在time.sleep(2)期间，程序完全阻塞在那一行，做不了任何其他事。
3.2 异步版示例，同样的IO操作几乎同时开始
改用asyncio来实现，只需要把time.sleep换成await asyncio.sleep，然后把调用放进事件循环：
import asyncio async def fetch_data(url): print(f"开始请求 {url}") await asyncio.sleep(2) print(f"结束请求 {url}") return f"数据 {url}" start = time.time() async def main(): tasks = [] for url in ["A.com", "B.com", "C.com", "D.com", "E.com"]: tasks.append(fetch_data(url)) await asyncio.gather(*tasks) asyncio.run(main()) print(f"耗时: {time.time() - start:.2f}秒")运行结果大概是耗时2秒。你如果观察输出，会发现5个“开始请求”几乎同时打出来，然后2秒后5个“结束请求”也几乎同时出现。这就是并发：每个协程都在等待时让出了CPU控制权，事件循环在它们之间飞快切换，最终把总耗时压到了一个等待周期。
3.3 为什么不能用time.sleep，这个坑很多人踩过
异步代码里面，严禁使用time.sleep。为什么？因为time.sleep会直接阻塞整个线程，事件循环也被卡住，所有其他正在等待的协程都没法继续执行。你以为是异步并发，结果又变成了同步串行，甚至比同步还慢——因为你还额外付了事件循环调度的开销。
正确做法是：需要模拟等待，用await asyncio.sleep()；如果是调用第三方同步库的耗时IO，要么用asyncio.to_thread()把它丢到线程池执行，要么换成支持异步的库（比如aiohttp、asyncpg）。后面我会专门讲这个问题的解决方案。
从这个简单的例子可以总结出异步程序的基本套路：把任务写成异步协程 → 在主协程里把所有协程包装成Task或直接用gather→ 用asyncio.run(主协程())启动整个流程。这个模板适用于绝大多数异步脚本。
4. 核心实操：并发控制、超时设置、任务取消
4.1 用gather收集多个结果，并处理异常
前面已经用过gather，但实际场景里我们经常会遇到某个任务出错，这个时候gather的行为需要了解一下。默认情况下，gather的多个协程里只要有一个抛异常，整个gather调用就会中断并抛出异常，其他协程即使已经运行也会被取消。想拿到每个任务的结果而不是被中断，可以设置return_exceptions=True：
async def safe_run(): results = await asyncio.gather( might_fail("任务1"), will_fail("任务2"), normal_task("任务3"), return_exceptions=True ) for r in results: if isinstance(r, Exception): print(f"任务失败: {r}") else: print(f"正常结果: {r}")设置成True之后，异常会作为结果返回，而不会中断其他任务。这在批量调用外部接口时非常实用，比如你要并发请求100个网页，其中3个超时或挂了，你希望剩下的97个照常工作，全部跑完再把失败的数据记下来重试。
4.2 用asyncio.wait控制并发数量
gather会把所有传入的协程一次性全部并发出去。如果任务是1000个，这1000个会同时跑。对于到外部API这种场景，有时候并不合适，因为对方服务器可能扛不住。这时候就需要限制最大并发数。
实现方式有很多种，最经典的是用信号量asyncio.Semaphore。它就像停车场入口的闸机，限定同一时间能通过几个车。
async def worker(url, sem): async with sem: print(f"开始请求 {url}") await asyncio.sleep(1) print(f"完成请求 {url}") return f"{url} 的结果" async def run_with_limit(urls, limit=3): sem = asyncio.Semaphore(limit) tasks = [worker(url, sem) for url in urls] results = await asyncio.gather(*tasks) return resultsSemaphore(limit)会让async with sem内部同一时间只有limit个协程能够进入。我这个例子设置了3，那么10个请求就会分批次执行，每次同时跑3个，完成一个才放进下一个。这个模式比多线程里手动维护队列要简洁得多，代码可读性也更好。
4.3 超时处理：不能无限等待一个挂死的请求
实际开发里最坑的场景就是：一个协程调外部接口，结果对方服务器一直不返回，你的程序就永远挂在await上。不设置超时的异步任务，在大型工程里就是定时炸弹。
Python 3.11以前，最常用的超时写法是用asyncio.wait_for：
try: result = await asyncio.wait_for(some_async_call(), timeout=5) except asyncio.TimeoutError: print("请求超时了")wait_for内部会在超时后触发取消（cancel）操作，停止那个协程。Python 3.11以后又新增了asyncio.timeout()上下文管理器，写法更优雅：
async with asyncio.timeout(5): result = await some_async_call()两段代码的功能类似，但wait_for在入门阶段更常用，文档也多，推荐新手先掌握它。记住：给每一个可能会等待的任务都设置超时，这是异步工程的第一生存法则。
4.4 任务取消：怎么优雅地停掉协程
有时候你不想等所有任务完成，比如用户点了取消按钮，或者发现某个条件不需要继续跑其他任务了。此时可以调用task.cancel()取消一个任务。取消后的事件循环会向该任务注入CancelledError异常。
在协程内部，要处理取消逻辑可以这样：
async def long_running(): try: await asyncio.sleep(3600) except asyncio.CancelledError: print("我被打断了，做一些清理") raise # 重新抛出，表示接受取消注意在except里，如果你的清理工作不需要再继续，最后一定记得raise把CancelledError重新抛出去，这样任务才会真正进入取消状态。如果你吞掉异常，任务会继续运行，可能造成逻辑混乱。
批量取消的场景可以用asyncio.gather(return_exceptions=True)配合循环取消所有任务：
tasks = [asyncio.create_task(fetch(url)) for url in urls] # 某个条件触发 for t in tasks: t.cancel() await asyncio.gather(*tasks, return_exceptions=True)4.5 在同步代码里调用异步：asyncio.run的边界
很多脚本一开头就是在普通函数里调用异步函数，很容易写出下面这种错误的代码：
def main(): asyncio.run(fetch_data()) # 第一次调用没问题 asyncio.run(fetch_data()) # 第二次调用会报错吗？其实asyncio.run每次都会创建一个新的事件循环，所以连续调用两次是没问题的。但如果是在已经运行的事件循环内部调用asyncio.run，就会报错。例如：
async def outer(): asyncio.run(inner()) # 这里会报错！原因是此时事件循环已经在运行，你不能再启动一个新的。解决方式是直接把inner()用await或者其他异步方式调用。这个坑提醒我们：要把程序入口规划好，尽量保持异步函数只被异步上层调用，把所有异步逻辑包在一个main()里，最后用asyncio.run(main())启动。
5. 实战进阶：把阻塞IO正确“异步化”
5.1 第三方库不支持async怎么办：使用to_thread
现实世界里，很多库（比如requests、pymysql）都是同步阻塞的。你不能直接在协程里调它们，否则会卡死整个事件循环。最简单的补救方案是asyncio.to_thread()，它会把这个函数扔到线程池里跑，同时把控制权还给事件循环。
来看一个例子，用requests并发下载数据：
import asyncio import requests async def download_with_requests(url): print(f"开始请求 {url}") # to_thread 把 requests.get 放到线程池执行 data = await asyncio.to_thread(requests.get, url, timeout=10) return data.text[:100] async def main(): urls = ["https://example.com"] * 5 tasks = [download_with_requests(url) for url in urls] results = await asyncio.gather(*tasks) print(f"拿到 {len(results)} 个结果") asyncio.run(main())注意requests.get本身是同步阻塞的，但通过to_thread把它放到其他线程运行，主线程的事件循环就不会被阻塞，其他异步协程依然可以并发推进。这算是一个“补救型”方案，适合快速改造现有代码。如果是在新项目里，更推荐直接用aiohttp这种原生支持异步的库，少一层线程开销，性能更好。
5.2 使用loop.run_in_executor实现更精细的线程池控制
to_thread底层用的就是run_in_executor，如果你需要控制线程池的大小，或者需要在线程池里跑一些有依赖的多个调用，可以用run_in_executor：
import asyncio from concurrent.futures import ThreadPoolExecutor async def main(): loop = asyncio.get_running_loop() executor = ThreadPoolExecutor(max_workers=4) with executor: result = await loop.run_in_executor(executor, requests.get, "https://example.com") print(result.status_code)run_in_executor的第一个参数是Executor实例，传None时默认使用默认的线程池。这种写法的好处是你可以针对特定任务自定义线程池，比如限制成2个线程，防止某个同步库自己有连接池上限导致卡顿。
5.3 一些容易忽略的细节
asyncio.to_thread在Python 3.9引入，3.9以下版本用run_in_executor代替。async with aiohttp.ClientSession()其实也是一种协程上下文管理器，使用方式和普通with不同，但不要搞混。- 如果你在协程里调用
time.perf_counter()这种纯CPU操作，不需要特殊处理，它们不影响事件循环。 - 文件读写虽然也是IO，但Python标准库的
open是同步的，如果要异步读文件，可以用aiofiles库。
6. 常见问题排查与经验技巧实录
6.1 报错：coroutine was never awaited
这个错误是最常见的入门报错，原因通常是忘记在协程前面加await，或者在创建Task时写错了。比如：
async def hello(): print("Hello") hello() # 不会执行，而且给出 RuntimeWarning解决方式：要么await hello()，要么asyncio.create_task(hello())。同时要注意，create_task只接受协程对象，如果你传入的是一个已经await得到的值，也会报错。
6.2 报错：There is no current event loop in thread
有时候你直接在主线程里用asyncio.get_event_loop()，会得到这个错误，尤其是在某些事件循环已经被关闭的情况下。现在官方推荐直接用asyncio.run()或asyncio.get_running_loop()（后者必须在事件循环内部使用）。不要再在普通函数里用get_event_loop()创建循环，容易踩雷。
6.3 异步代码里不小心调用了阻塞操作
比如在协程里直接写time.sleep(2)，或者调用requests.get，表现出来就是整个程序时长没有减少，甚至卡住。排查方法很简单：在异步代码里搜索关键词time.sleep和requests.，如果是time.sleep，换成await asyncio.sleep；如果是requests，换成aiohttp或to_thread。
6.4 并发数量过高导致拒绝服务
即使异步并发性能再好，也不要无脑把所有任务一股脑丢给gather。我见过一个事例：某人同时放了5000个爬虫任务去请求一个小网站，网站直接熔断，反爬虫策略接踵而至。实际工程里，务必用Semaphore限制并发，通常设置成50~200之间，具体看目标服务器的承受能力和网络状况。
6.5 忘记设置任务超时导致进程永远挂起
如果你没有给任务设置超时，外部请求一直不返回，进程就像死了一样。解决方式参照上面的wait_for。一个小技巧：可以在gather外面再包一层wait_for，给整体操作设一个最大时间：
try: await asyncio.wait_for(asyncio.gather(*tasks), timeout=30) except asyncio.TimeoutError: print("整体超时，强制取消所有任务")这样即使个别任务内部没有超时保护，整个流程不会无限期卡住。
6.6 调试技巧：打印事件循环中的任务
排查性能问题时，可以在适当位置打印当前事件循环里有多少个任务在运行：
async def debug_tasks(): loop = asyncio.get_running_loop() for task in asyncio.all_tasks(loop): print(f"任务: {task}") print(f"状态: {task._state}")在长时间运行的异步服务里，这个技巧能帮你快速找出积压的任务、卡死的协程，定位到底哪一步没有返回。
6.7 与多线程配合时的注意事项
asyncio虽然自带线程池（to_thread），但线程和事件循环之间传递数据有讲究。不要在非异步线程里直接调用async函数，应该使用asyncio.run_coroutine_threadsafe()：
from concurrent.futures import ThreadPoolExecutor import asyncio async def async_task(): return 42 loop = asyncio.new_event_loop() thread_pool = ThreadPoolExecutor(max_workers=1) def run_loop(): asyncio.set_event_loop(loop) loop.run_forever() thread_pool.submit(run_loop) # 在另一个线程里，安全地把协程扔进事件循环 future = asyncio.run_coroutine_threadsafe(async_task(), loop) result = future.result()run_coroutine_threadsafe会返回一个concurrent.futures.Future，可以用future.result()同步等待异步结果。这个方法用在多线程GUI程序、ORM回调里非常常见，建议先收藏，用到时再细看。
7. 一个综合示例：并发请求API并保存结果
把所有要点串起来，写一个小工具：并发请求某个接口获取用户信息，限定最大并发5个，每个请求超时10秒，失败的重试一次，最终把成功的数据写入JSON文件。这个例子的代码可以直接改改URL用到你的场景里。
import asyncio import json import time import aiohttp async def fetch_user(session, user_id, sem): async with sem: url = f"https://api.example.com/users/{user_id}" for attempt in range(2): # 最多重试1次 try: async with session.get(url, timeout=10) as resp: if resp.status == 200: data = await resp.json() print(f"成功获取 {user_id}, 第{attempt+1}次") return data else: print(f"用户 {user_id} HTTP状态码 {resp.status}") except asyncio.TimeoutError: print(f"用户 {user_id} 超时") except aiohttp.ClientError as e: print(f"用户 {user_id} 请求错误: {e}") await asyncio.sleep(1) # 重试前等一下 return None async def main(): sem = asyncio.Semaphore(5) timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: tasks = [fetch_user(session, uid, sem) for uid in range(1, 101)] results = await asyncio.gather(*tasks) success = [r for r in results if r is not None] print(f"成功: {len(success)} / 总数: {len(results)}") with open("users.json", "w", encoding="utf-8") as f: json.dump(success, f, ensure_ascii=False, indent=2) if __name__ == "__main__": asyncio.run(main())这个示例里，Semaphore限制了同时发起的请求数，ClientTimeout设置了整体超时，协程内部又有尝试计数和异常分类。100个请求，并发5个，就算每个请求50毫秒，整体也只需要2秒左右，而同步遍历100个请求至少需要好几秒甚至几十秒。异步在这里的优势被体现得淋漓尽致。
想真正用好asyncio，一定要亲手跑一遍上面的代码，改一改数字，体会一下事件循环的调度。我的个人建议是：先不要碰那些花哨的第三方异步框架，把asyncio内置的run、gather、create_task、Semaphore、wait_for用熟，能覆盖日常开发中90%的异步场景。等熟悉了这几个工具，再去读Streams、Protocol和Transport就不会觉得吃力。
最后分享一个小技巧：在调试异步爬虫时，我习惯在关键协程的入口和出口各打一条日志，并且打印出当前时间，比如print(f"[{time.strftime('%H:%M:%S')}] start {user_id}")。因为异步程序的执行顺序不像同步代码那样容易一眼看穿，加日志能帮你准确知道每个任务是什么时候开始、什么时候结束，排查并发问题会轻松非常多。异步编程不是一个看一遍就能掌握的东西，但只要照着这篇文章的代码自己动手改几遍，你会慢慢找到那种“所有任务为我同时运转”的快感。
