Python asyncio 数据爬取：并发 100 个请求的正确姿势

Python asyncio 数据爬取并发 100 个请求的正确姿势一、为什么多线程在爬虫场景里不够用大家好我是朱大喜。作为数据分析师免不了要写爬虫抓数据——竞品价格、公开财报、行业资讯这些事情在数据组里都是家常便饭。很多同学一开始用requestsfor循环一次一个 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, timeoutaiohttp.ClientTimeout(total30)) 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(limitconcurrency 20, limit_per_hostconcurrency) async with aiohttp.ClientSession(connectorconnector) as session: # asyncio.gather 把所有协程聚合成一个等待全部完成 tasks [bounded_fetch(session, url) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) return results # 使用示例 async def main(): urls [fhttps://api.example.com/item/{i} for i in range(100)] start time.time() results await fetch_concurrent(urls, concurrency100) elapsed time.time() - start print(f100个请求并发100耗时: {elapsed:.2f}秒) # Python 3.7 推荐用法 if __name__ __main__: asyncio.run(main())这里有几个容易踩坑的设置单独拎出来说Semaphore信号量是必须的。不要以为开 100 个并发就设 concurrency100 就行了——如果目标网站是你自己的内部 API100 并发可能没事但如果是外部网站100 并发等于 DDoS 攻击封 IP 是分分钟的事。一般公开网站建议 10-20内部 API 可以到 50-100根据服务器承受能力调整。TCPConnector 的 limit 要设得比信号量略大。因为连接建立、SSL 握手也有开销如果 limit 刚好等于并发数当连接池满了而又有新请求时可能会多一个等待时间。设大一点20能避免这个边界问题。asyncio.gather 的 return_exceptions 参数。默认情况下任何一个协程抛出异常gather 会立即抛出正在跑的协程会被取消。设return_exceptionsTrue后异常会被当作结果返回而不是中断整个批次。对于爬虫来说少数几个 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, timeoutaiohttp.ClientTimeout(total30)) 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。这大概就是它最迷人的地方。