企业官网建设流程全解析

写爬虫的时候，你大概率遇到过这种场景：同步代码发几百个请求，每个都要等网络返回，跑完一轮要十几分钟；换线程池倒是快了，但线程切换、共享变量加锁、回调嵌套，写着写着就变成一团乱麻。这其实就是Python异步编程要解决的问题——让程序在等待IO的间隙去做别的事，而不是干等。作为Python官方自带的异步方案，Asyncio库是我近两年用得最多的工具，没有之一。这篇文章不聊复杂理论，我从同步代码为什么会慢讲起，把事件循环、协程、Task、并发限制、超时处理逐个拆开，最后附上我自己踩过的坑，争取让没用过异步的人也能照着把代码跑起来。

1. 从“等”到“不等”：异步真正解决的问题

1.1 同步代码的时间账：串行等待的代价

先看一段典型的同步代码：用requests逐个请求十个接口，每个接口响应200毫秒。十个请求串行执行，总耗时就是10 × 0.2秒等于2秒。听起来不多，但如果换成抓取一千个商品详情页，每个页面还是200毫秒，那就是200秒，整整三分多钟。程序在这三分钟里几乎全在“等网络”，CPU闲着，磁盘闲着，任务列表却一个接一个排着队。

这时候第一反应是开线程池。线程池确实能把十个请求并行起来，总耗时降到接近单个请求的时间，但代价是：线程创建和切换有开销、多个线程同时写一个变量要考虑锁、回调函数一层套一层，代码很快变得难维护。而且Python的GIL决定了线程在CPU密集场景下也占不到便宜。相比之下，异步的思路完全不同——它不是把任务拆给多个线程，而是让单个线程在“等待”的时候主动去执行别的任务，整体看起来就像多任务同时进行。

1.2 事件循环是怎么“插队”的

Asyncio的核心是一个事件循环，你可以把它理解成一张待办清单。程序把所有协程任务扔进清单，事件循环从第一个任务开始执行；协程执行到await关键字时，表示“我要等一个IO结果，暂时干不了别的”，事件循环立刻把这个任务放到一边，转去执行清单里的下一个任务；等IO结果返回了，再把任务放回清单继续跑。

还是拿上面的请求举例。A协程发出请求后等待网络响应时会说“我歇了”，事件循环转去执行B协程，B发出请求后也歇了，接着轮C……十个协程全部发出请求并进入等待状态，然后第一个响应回来了，事件循环再接着处理A的剩余逻辑。整个过程只有一个线程，却能达到近似并行的效果，因为网络IO的等待时间被完全重叠了。这个机制和现实中前台点单很像：咖啡师同时接十几单，做A美式的时候等着浓缩萃取，转身去把B拿铁的牛奶打了，谁先准备好就先做谁，而不是一杯做完再做下一杯。

需要提醒的是，异步擅长的是“等待型”任务，也就是网络请求、文件读写、数据库访问这类以IO等待为主的操作。如果是纯CPU计算，比如大量数值运算、图片滤镜处理，异步不但不会更快，反而会因为事件循环的调度开销变慢，这种情况更适合多进程。

1.3 什么时候该用异步，什么时候别硬上

我的经验是三个字：看瓶颈。程序运行慢，先问一句“时间花在哪了”。如果是花在网络IO或磁盘读写上，异步能带来肉眼可见的提升；如果是花在CPU计算上，异步帮不上忙。

适合用Asyncio的场景很明确：爬虫和API客户端、Web服务端（比如FastAPI这类异步框架）、消息队列的消费者、需要同时监控大量网络连接的场景。不适合的场景也很明确：计算密集型的脚本、单次调用就能搞定的简单任务、以及团队还没人能看懂异步语法的情况下硬改造老代码。

另外有个务实建议：别把项目里所有同步代码都推倒重写。异步代码有个特点，它内部的同步阻塞调用会“卡死”整个事件循环，所以一个项目里最好保持“全异步”或“全同步”，混着用反而容易出事。我在第五节会细讲这个坑。

2. 入门Asyncio：协程、事件循环与任务三件套

2.1 协程函数和await的语法地基

Python里用async def定义的函数叫协程函数，调用它不会立即执行，而是返回一个协程对象。只有把协程对象交给事件循环，里面的代码才会真正跑。协程函数内部，用await等待另一个协程或Future对象；执行到await时，当前协程挂起，事件循环去处理其他任务。

import asyncio async def say_after(delay, word): print(f"开始: {word}") await asyncio.sleep(delay) print(f"结束: {word}") async def main(): print("main 开始") await say_after(1, "hello") await say_after(2, "world") print("main 结束") asyncio.run(main())

上面这段代码里，第一个await会让事件循环干等1秒，然后再执行第二个。两个协程串行执行，总耗时3秒。注意asyncio.sleep不是time.sleep，前者是异步的，挂起时事件循环还能干别的；后者是阻塞的，一睡下去整个事件循环都跟着睡。

2.2 创建任务的三种姿势：怎么让协程真正并发

只写await并不一定并发。想让多个协程同时跑，得把它们包装成Task。Task可以理解为“事件循环里的一个待办事项”，创建Task的那一刻，协程就开始蓄力运行了。

最常用的是asyncio.create_task()，在已有的事件循环内创建任务：

async def main(): task1 = asyncio.create_task(say_after(1, "hello")) task2 = asyncio.create_task(say_after(2, "world")) await task1 await task2

这段代码里两个协程并发，总耗时约2秒。asyncio.gather()是另一个常用入口，它同时聚合多个协程并等待全部结果返回，返回顺序和传入顺序一致：

results = await asyncio.gather( say_after(1, "hello"), say_after(2, "world"), )

Python 3.11以后还推荐用asyncio.TaskGroup，它比裸用gather更好的一点是：组里任何一个任务抛异常，其他任务会被自动取消，避免出现“一个任务挂了，其他任务还在后台空转”的情况：

async def main(): async with asyncio.TaskGroup() as tg: tg.create_task(say_after(1, "hello")) tg.create_task(say_after(2, "world"))

对于日常写脚本，我最常用的还是create_task加gather组合：分批创建任务，再统一收结果。这个模式最直观，也最容易控制并发数量。

2.3 运行入口的写法：asyncio.run为什么是首选

旧的教程里经常出现loop = asyncio.get_event_loop()然后loop.run_until_complete(...)的写法，我建议你直接跳过，用asyncio.run()就够。asyncio.run()做了三件事：创建事件循环、运行传入的协程直到结束、关闭事件循环并清理资源。它还会自动处理一些旧写法容易遗漏的清理问题，比如关闭异步生成器。

import asyncio async def main(): print("跑起来了") if __name__ == "__main__": asyncio.run(main())

有一点要注意：asyncio.run()每次调用都会创建一个新的事件循环，所以不能在同一个事件循环里多次调用它，也不能在一个已经运行的协程内部再去调用asyncio.run()。项目里事件循环的唯一入口就是main协程，逻辑都往里面放。

3. 实操：用Asyncio写一个并发下载脚本

3.1 选型：为什么请求库选aiohttp而不是requests

既然目标是异步，那HTTP请求也得是异步的。requests是同步库，在协程里调用它会阻塞事件循环，等于把异步的优势全丢了。aiohttp是目前最成熟的异步HTTP库，提供了ClientSession和session.get()这类和requests用法相似的API，上手成本低。

import asyncio import aiohttp async def fetch(session, url): async with session.get(url) as resp: return await resp.text() async def main(): urls = [ f"https://example.com/api/data/{i}" for i in range(10) ] async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(fetch(session, url)) for url in urls] pages = await asyncio.gather(*tasks) print(f"抓取完成，共 {len(pages)} 个页面") asyncio.run(main())

这段代码的含义：先创建一个全局的ClientSession，它内部会维护连接复用，避免每个请求都重新建连；然后为每个URL创建一个Task，gather等待全部完成。注意这里和第2节的串行例子有个关键区别——所有session.get()发出后，事件循环会同时等待它们的响应，而不是等一个完成再发下一个。上面的example.com只是为了演示结构，实际跑的时候替换成自己的接口地址就行。

3.2 并发控制：防止把对方接口打崩

直接把几百个URL全部create_task，一瞬间发出几百个并发请求，大概率会被对方服务拒掉，自己的机器也会因为文件描述符不够报错。业内普遍的做法是加信号量限制并发数量。asyncio.Semaphore的原理很简单：内部有个计数器，进入时减一，离开时加一，计数器为零时后续协程只能等待。

import asyncio import aiohttp SEMAPHORE_LIMIT = 5 async def fetch(session, sem, url): async with sem: async with session.get(url) as resp: return await resp.text() async def main(): urls = [f"https://example.com/api/data/{i}" for i in range(30)] sem = asyncio.Semaphore(SEMAPHORE_LIMIT) async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(fetch(session, sem, url)) for url in urls] results = await asyncio.gather(*tasks) print(len(results)) asyncio.run(main())

我习惯把并发上限设成一个常量，比如5到10。它解决的不只是“不要打死别人服务器”的问题，更是让自己程序稳定：并发太高时，TCP连接数、内存占用、超时重试都会变得不可控。调试阶段先用限到3跑通，再逐步调大，这是我在实践中觉得最稳妥的节奏。另外注意信号量对象要传进协程里，而不是在协程内部创建——如果在每个协程里都创建一个新Semaphore，那就等于没有限制。

3.3 结果对比：异步到底快了多少

我拿一个模拟场景做过粗略对比：10个请求，每个请求等待200毫秒。同步版本串行耗时约2秒；线程池5个线程版本约0.4秒，但线程切换和锁竞争会让代码复杂度上升；异步版本用Semaphore限到5，耗时约0.4秒，而代码结构却是线性书写的。当请求数量涨到100个时，异步版本的优势会更明显，因为线程池受线程数限制，大量任务仍在排队，而异步的事件循环调度成本远低于线程切换成本。

让我用一张表说明这个关系：

需要说明的是，这里的数字是我在自己机器上的粗略测试，真实场景受网络波动影响很大。但趋势是确定的：IO密集任务里，异步在并发规模越大时越有优势，而且写出可控的并发策略比线程锁要省心得多。

4. 复杂场景的进阶处理：超时、限流与任务编排

4.1 给耗时操作加上超时保护

现实中的网络请求没有理想的200毫秒。服务端响应变慢、DNS解析卡住、连接被重置，任何一个毛病都可能让协程挂在那里半天不返回。所以给每个协程加超时是我写异步代码的基本要求。

asyncio.wait_for()是最直接的工具，它给一个协程设置时间上限，超时后内部会尝试取消该协程，并抛出asyncio.TimeoutError：

import asyncio async def slow_op(): await asyncio.sleep(10) return "结果" async def main(): try: result = await asyncio.wait_for(slow_op(), timeout=3) print(result) except asyncio.TimeoutError: print("操作超时，已取消")

超时的本质是“取消任务”。如果一个协程内部有需要清理的资源，比如已打开的连接、临时文件，那被取消时就要处理asyncio.CancelledError，确保资源释放干净。第五节我会专门讲取消的坑。用了wait_for之后，程序的鲁棒性会明显提升：一个请求超时不会再让整个任务组卡死。

在aiohttp场景里，我更推荐同时设置连接超时和整体读取超时。ClientTimeout可以分别控制连接、读取、写入的时间：

timeout = aiohttp.ClientTimeout(total=10, connect=5) async with aiohttp.ClientSession(timeout=timeout) as session: ...

我的习惯是整体总时长设到10秒，连接阶段5秒，这样即使某个页面卡死，也最多拖10秒就重试或跳过，不会把整个任务拖垮。

4.2 批量任务的分组与重试策略

并发抓取几百个URL时，一次性gather全部任务不合适。一方面是内存里堆着几百个任务对象和结果字符串，占用大；另一方面是某几个请求失败时，不知道是重试还是放弃。

目前我常用的模式是“分批 + 收集失败项”。把URL列表按每批20个切片，每批内部并发，跑完后检查结果，失败的URL单独记录，最后统一重试。这样既控制了并发峰值，又让逻辑清晰：

BATCH_SIZE = 20 async def run_batches(session, sem, urls): failed = [] for i in range(0, len(urls), BATCH_SIZE): batch = urls[i:i + BATCH_SIZE] tasks = [asyncio.create_task(fetch(session, sem, url)) for url in batch] results = await asyncio.gather(*tasks, return_exceptions=True) for url, result in zip(batch, results): if isinstance(result, Exception): failed.append(url) return failed

这里有个小技巧：gather(..., return_exceptions=True)不会让单个任务的异常直接打破整个批次，而是把异常当作返回值收集起来。随后遍历批次，把异常项对应到URL，方便重试。这个模式我几乎在每类异步采集脚本里都会复用。

4.3 回调还是等待：和Future打交道的方式

协程内部偶尔会遇到“某个事件由其他任务触发”的场景，最常见的做法是使用asyncio.Event，它的用法和线程版Event类似，只是wait()是异步的。两个或多个任务可以用它互相告知状态，而不是靠轮询检查变量。

import asyncio async def waiter(event): print("等待事件触发...") await event.wait() print("事件已触发，继续执行") async def setter(event): await asyncio.sleep(2) event.set() async def main(): event = asyncio.Event() asyncio.create_task(waiter(event)) asyncio.create_task(setter(event)) await asyncio.sleep(3) asyncio.run(main())

有人会问：这里为什么不能用time.sleep加一个全局变量轮询？因为time.sleep是阻塞的，一旦在任意一个协程里调用，整个事件循环的调度都停摆，其他任务全部卡住。只要记住一点：在异步世界里，凡是“等待”都要用异步版本的函数，比如asyncio.sleep、asyncio.Event.wait()、aiohttp的请求，而不是它们的同步等价物。

5. 实战中最容易踩的五个坑

5.1 在协程里写阻塞代码，等于自断经脉

这是异步脚本最常见的性能杀手。我见过不少把异步爬虫写完发现速度没提升的例子，最后定位到原因：协程里调用了requests.get()，或者用了time.sleep()。这两个调用都是阻塞的，执行时事件循环没法切换去处理其他任务，整个并发就名存实亡。

判断标准很简单：如果一个库的函数会真正“等”某件事完成，且它没有异步版本，那它就不该出现在协程内部。如果实在绕不开某个同步库，可以用loop.run_in_executor()把它扔到线程池里执行，让事件循环不被阻塞。建议在写协程之前就锁死规矩：统一用aiohttp、asyncpg、aiofiles这类异步库，不混用同步IO。

5.2 忘记await导致的事件循环警告

新手经常写出这样的代码：调用了协程函数，但忘了await，结果得到一个coroutine对象，函数体根本没执行。更隐蔽的问题是，创建了Task却忘了引用它，导致任务被垃圾回收或者产生“未等待的协程”警告。

# 错误示例：忘了 await async def main(): coro = say_after(1, "hello") # 此时函数还没有执行 # 这个协程对象会被丢弃，Python 会发出 RuntimeWarning

解决方案是养成两个习惯：一是协程函数调用后要么立即await，要么立刻包成Task；二是在代码开头加上调试模式，让未等待的协程直接报warning而不是静默吞掉：

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

出现问题时，调试模式下事件循环会打印更详细的调度信息，对新手排查“为什么我的协程没跑”很有帮助。

5.3 任务取消与CancelledError的处理

asyncio.CancelledError在Python 3.8之后继承自BaseException，而不是Exception。这意味着如果在except Exception里写清理逻辑，取消异常不会被捕获，很多清理代码等于白写。正确做法是单独用except asyncio.CancelledError处理，或者用finally保证资源释放：

async def worker(): try: await asyncio.sleep(10) except asyncio.CancelledError: # 清理连接、关闭文件等收尾工作 raise # 重新抛出，通知上层任务确实被取消了 finally: await cleanup()

有个细节值得记一下：任务被取消后，在except块里做清理时，如果清理逻辑里又有耗时的await，这些操作要尽量轻量。因为一个任务在取消路径中“垂死挣扎”的时间也被算进wait_for超时里，清理太重会拖慢整体取消。

5.4 数据库和Redis的驱动选择

把异步爬虫跑通之后，很多人接着要把数据写进数据库，然后发现整个程序又卡了。原因大概率是用了pymysql、psycopg2这类同步驱动，数据库查询的等待时间把事件循环堵住了。异步项目要用异步驱动，比如asyncpg配PostgreSQL，aiomysql配MySQL，aioredis（现在并入了redis.asyncio）配Redis。它们的API风格和同步版有点差异，但熟悉之后差别不大。

我踩过的具体坑：数据量不大，但每写一条记录都要重新建立一次连接。后来把所有写入操作统一在一个连接池里复用，整体耗时从几十秒降到几秒。记住异步编程里“连接复用”不是优化技巧，而是基本功。

5.5 日志排障的笨办法：把任务ID打出来

异步代码的报错经常让人懵，因为多个任务交错执行，日志顺序毫无规律，你不知道某条错误对应哪个请求。我的做法很简单：每个任务都带一个可识别的ID，日志里打印“任务ID + 当前状态”，例如“task-3 开始请求”，出现异常时也能直接定位到具体URL。

还可以利用asyncio.all_tasks()在异常时打印当前所有存活任务，快速判断是不是有任务被卡死没退出。这个方法虽然笨，但在没有上监控工具之前，是排查异步程序卡死最高效的手段。

写到这里，最后分享一点个人体会。刚接触Asyncio那阵子，我一度追求“把一切代码都改异步”，结果多个同步库混用，程序不但没变快，反而频繁卡住。后来悟出两个道理：一是异步的收益来自于IO密集且并发规模大的场景，别为了异步而异步；二是异步代码的调试成本高于同步代码，前期多花时间写清楚日志、控制好并发上限，比后期排查不明异常省力得多。如果你刚上手，不妨先别管那些花哨特性，就从学会创建Task、用Semaphore限流、给请求加超时这三件事开始，跑通一个真实的小脚本，再逐步往深里走。