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。这大概就是它最迷人的地方。