Python asyncio 数据爬取：并发 100 个请求的正确姿势
一、为什么多线程在爬虫场景里不够用
大家好，我是朱大喜。作为数据分析师，免不了要写爬虫抓数据——竞品价格、公开财报、行业资讯，这些事情在数据组里都是家常便饭。
很多同学一开始用 requests + for 循环，一次一个 URL 地请求。100 个页面等 100 秒，慢得让人想泡杯茶回来再看。然后学会了 ThreadPoolExecutor，开了 10 个线程，100 个页面 10 秒搞定，感觉已经很快了。
但遇到 10000 个页面的时候呢？线程的上下文切换开销、GIL 锁的争夺、线程池创建和销毁的成本，这些都开始变得不可忽略。更要命的是，HTTP 请求的本质是 IO 等待，不是 CPU 计算。你开了 50 个线程，大部分时间它们都在干等服务器响应，CPU 根本没跑满。
这时候 asyncio 就该登场了。asyncio 的核心思想非常直白：在等待 IO 的时候切换去做别的事。它像一个高效的奶茶店店员，正在等热水烧开的时候不会干站着，而是去加糖、封杯、叫号——一个人同时处理多个订单。
二、asyncio 的正确打开方式
asyncio 不是魔法，它有严格的适用条件。你用的 HTTP 库必须支持异步——requests 库是同步的，放在 async 函数里调用不仅不加速，反而会阻塞整个事件循环。正确的选择是 aiohttp 或者 httpx 的 async 模式。
import asyncio
import aiohttp
import time
from typing import List, Dict, Any
async def fetch_one(session: aiohttp.ClientSession, url: str) -> Dict[str, Any]:
"""
异步抓取单个URL的数据
为什么用 async/await？因为 HTTP 请求的等待时间不耗CPU，
这段时间可以让给其他协程去跑
"""
try:
# async with 确保请求完成后自动关闭连接，防止连接泄漏
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
# 等待服务器返回时，控制权自动让出给事件循环
data = await resp.json()
return {"url": url, "status": resp.status, "data": data}
except asyncio.TimeoutError:
return {"url": url, "status": "timeout", "data": None}
except Exception as e:
return {"url": url, "status": "error", "error": str(e)}
async def fetch_concurrent(urls: List[str], concurrency: int = 100) -> List[Dict]:
"""
并发抓取多个URL，使用信号量控制并发数
concurrency: 最大同时请求数，避免把目标服务器打爆
"""
# 信号量就像一个停车场的闸机：
# 停车场有N个车位（concurrency），满了就排队等
semaphore = asyncio.Semaphore(concurrency)
async def bounded_fetch(session, url):
# async with semaphore 确保同一时间最多只有 concurrency 个协程在跑
async with semaphore:
return await fetch_one(session, url)
# TCPConnector 控制连接池大小，limit 略大于 concurrency 避免连接不够用
connector = aiohttp.TCPConnector(limit=concurrency + 20, limit_per_host=concurrency)
async with aiohttp.ClientSession(connector=connector) as session:
# asyncio.gather 把所有协程聚合成一个，等待全部完成
tasks = [bounded_fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 使用示例
async def main():
urls = [f"https://api.example.com/item/{i}" for i in range(100)]
start = time.time()
results = await fetch_concurrent(urls, concurrency=100)
elapsed = time.time() – start
print(f"100个请求，并发100，耗时: {elapsed:.2f}秒")
# Python 3.7+ 推荐用法
if __name__ == "__main__":
asyncio.run(main())
这里有几个容易踩坑的设置，单独拎出来说：
Semaphore（信号量）是必须的。不要以为开 100 个并发就设 concurrency=100 就行了——如果目标网站是你自己的内部 API，100 并发可能没事；但如果是外部网站，100 并发等于 DDoS 攻击，封 IP 是分分钟的事。一般公开网站建议 10-20，内部 API 可以到 50-100，根据服务器承受能力调整。
TCPConnector 的 limit 要设得比信号量略大。因为连接建立、SSL 握手也有开销，如果 limit 刚好等于并发数，当连接池满了而又有新请求时，可能会多一个等待时间。设大一点（+20）能避免这个边界问题。
asyncio.gather 的 return_exceptions 参数。默认情况下，任何一个协程抛出异常，gather 会立即抛出，正在跑的协程会被取消。设 return_exceptions=True 后，异常会被当作结果返回而不是中断整个批次。对于爬虫来说，少数几个 URL 失败不该影响大局。
三、控制并发节奏：别把服务器打哭了
并发不是越多越好。我们用一个实际的对比来感受一下：
graph LR
subgraph "同步请求 (requests)"
A1[请求1: 2s等待] –> A2[请求2: 2s等待] –> A3[请求3: 2s等待]
A3 –> A4["…共100个"]
A4 –> A5[总耗时: ~200s]
end
subgraph "asyncio 100并发"
B1[请求1-100同时发出]
B1 –> B2[等待2s后全部返回]
B2 –> B3[总耗时: ~2s]
end
style A5 fill:#f44336,color:#fff
style B3 fill:#4CAF50,color:#fff
但有经验的爬虫工程师都知道一个残酷的现实：你并发开太高，对方服务器要么限流，要么直接封你。所以实际的正规爬虫，需要在这些基础上再加几层防护：
1. 随机延迟：每个请求之间加上 asyncio.sleep(random.uniform(0.5, 2.0))，模拟人类浏览行为。
2. 失败重试 + 退避：请求失败后不要立即重试，而是指数级延长等待时间。第一次失败等 1 秒，第二次等 2 秒，第三次等 4 秒……这样既给服务器喘息的机会，也不会因为瞬时错误放弃整个任务。
import random
async def fetch_with_retry(
session: aiohttp.ClientSession,
url: str,
max_retries: int = 3
) -> Dict[str, Any]:
"""
带指数退避的重试逻辑
原理：每次失败后等待时间翻倍，避免"雪崩式"重试把对方打得更惨
"""
for attempt in range(max_retries):
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
if resp.status == 429:
# 429 Too Many Requests: 被限流了，等久一点再试
retry_after = int(resp.headers.get("Retry-After", 2 ** attempt))
await asyncio.sleep(retry_after)
continue
return {"url": url, "status": resp.status, "data": await resp.json()}
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == max_retries – 1:
return {"url": url, "status": "error", "error": str(e)}
# 指数退避：1秒 -> 2秒 -> 4秒
wait_time = 2 ** attempt + random.uniform(0, 1)
await asyncio.sleep(wait_time)
return {"url": url, "status": "max_retries_exceeded"}
3. 请求头的诚意：加上合理的 User-Agent，尊重 robots.txt。这是职业操守，也是减少被封概率的最简单方式。
四、进阶：流式处理大批量 URL
如果 URL 数量是百万级的，一上来就全部加载到内存里显然不现实。这时候需要生产者-消费者模式：一个协程负责生产 URL（读文件/查数据库），多个协程负责消费（发送请求），中间用 asyncio.Queue 做缓冲。
async def producer(queue: asyncio.Queue, urls: List[str]):
"""生产者：往队列里放URL"""
for url in urls:
await queue.put(url) # 队列满了会阻塞，避免内存爆炸
# 放 N 个 None 作为结束信号（N = 消费者数量）
for _ in range(NUM_WORKERS):
await queue.put(None)
async def consumer(queue: asyncio.Queue, session: aiohttp.ClientSession, worker_id: int):
"""消费者：从队列取出URL并发请求"""
while True:
url = await queue.get()
if url is None: # 收到结束信号，退出循环
queue.task_done()
break
try:
result = await fetch_one(session, url)
# 这里可以把结果写入数据库或文件，而不是堆在内存里
await save_result(result)
finally:
queue.task_done() # 告诉队列：这个任务我处理完了
生产者-消费者模式的好处是背压控制：Queue 有个 maxsize 参数，当队列满了，producer 的 put 操作会被阻塞，从而自然地把 URL 的生产速度降下来。这避免了内存被 URL 列表撑爆的问题。
五、总结
asyncio 在数据爬取场景里确实很香，但用好它有几个硬性要求：第一，记得用 aiohttp 而不是 requests；第二，Semaphore 控制并发上限，别裸奔；第三，失败重试带上指数退避，尊重目标服务器；第四，大批量数据用生产者-消费者模式做流式处理。
很多初学者觉得 asyncio 难，主要是因为 async/await 的传染性——一个函数变成了 async，调用它的函数也得变成 async，一路往上传染到入口。但一旦你习惯了这种写法，回头看多线程的 ThreadPoolExecutor，反而会觉得那才是真正的复杂——各种锁、竞态条件、GIL，心累。
async 的世界里没有锁，只有 await。这大概就是它最迷人的地方。
