如果你写过爬虫一定经历过这样的场景明明只需要抓取1000个网页程序却跑了整整半个小时CPU占用率不到10%网络带宽也远未跑满可进度条就是慢吞吞地往前挪。问题出在哪里答案很简单——你的爬虫在大部分时间里都在“傻等”。传统的同步爬虫基于requests库发送HTTP请求后线程会阻塞直到服务器返回响应才能继续执行下一个请求。网络延迟、服务器处理时间、TCP握手开销这些等待时间加起来往往占了总耗时的90%以上。换句话说你的爬虫90%的时间都在“摸鱼”而不是真正在工作。异步爬虫正是为了解决这个问题而生。通过asyncio和aiohttp我们可以让程序在等待一个请求响应的同时发起另一个请求从而将等待时间重叠起来实现并发请求。理论上如果有N个请求需要执行同步爬虫耗时约为N * TT为单个请求平均耗时而异步爬虫理想情况下可以接近T假设并发数足够大且服务器不限制。在实际项目中将爬取效率提升5到10倍是非常常见的。本文将从零开始带你构建一个完整的异步爬虫项目。我们会先对比同步与异步的性能差异再深入aiohttp的核心用法最后封装一个可复用的异步爬虫框架。阅读本文需要你有Python基础熟悉基本的爬虫概念如HTTP请求、HTML解析但不需要你精通异步编程——我会把每个概念都讲清楚。目录第一章同步爬虫的痛点与异步的优势1.1 同步爬虫的工作模型1.2 多线程/多进程能解决问题吗1.3 异步IO模型事件循环与协程1.4 异步爬虫的性能潜力第二章aiohttp基础——异步HTTP客户端2.1 为什么选择aiohttp而非requests2.2 安装与环境准备2.3 ClientSession会话管理的核心2.4 发起GET/POST请求2.5 超时设置与异常处理2.6 连接池与并发控制第三章构建第一个异步爬虫3.1 目标抓取一个新闻网站的标题列表3.2 异步请求与HTML解析结合3.3 性能对比实验第四章异步爬虫进阶——错误重试与动态并发4.1 优雅的重试机制4.2 动态调整并发数自适应流控4.3 使用asyncio.Queue实现生产者-消费者模式第五章进阶技巧——Headers伪装、代理与Cookie持久化5.1 模拟浏览器Headers5.2 使用代理Proxy5.3 Cookie持久化与自动管理第六章实战项目——异步爬取百万级商品数据架构设计6.1 项目需求6.2 架构分层设计6.3 关键代码片段6.4 监控与日志6.5 断点续爬实现第七章性能调优与常见坑7.1 连接池大小与并发数调优7.2 防止DNS缓存污染7.3 内存泄漏防范7.4 事件循环的优雅关闭第八章异步爬虫的局限性及替代方案8.1 什么时候不适合用异步爬虫8.2 与Scrapy框架的对比8.3 未来趋势结语异步思维改变你的爬虫开发方式附录完整项目代码示例可直接运行第一章同步爬虫的痛点与异步的优势1.1 同步爬虫的工作模型我们先看一段典型的同步爬虫代码pythonimport requests import time urls [https://httpbin.org/delay/1 for _ in range(10)] # 模拟10个延迟1秒的请求 def sync_fetch(urls): results [] for url in urls: resp requests.get(url) results.append(resp.json()) return results start time.time() sync_fetch(urls) print(f同步耗时: {time.time() - start:.2f}秒)运行这段代码你会看到耗时大约10秒。每个请求发起后程序就卡在requests.get()那里等待服务器响应httpbin.org/delay/1会故意等待1秒再返回。10个请求串行执行总耗时就是10秒加上网络开销。问题本质IO密集型任务网络请求占据了绝大部分时间但CPU却在阻塞期间闲置。同步模型无法利用等待时间去执行其他任务。1.2 多线程/多进程能解决问题吗有人会说“那我用多线程不就行了”确实多线程可以实现并发请求但存在几个问题线程开销创建和销毁线程有成本并发数太高时如上千个系统资源消耗巨大。GIL限制Python的全局解释器锁使得多线程在CPU密集型任务上无效但IO密集型任务中线程切换仍有开销。代码复杂度线程间共享状态需要加锁容易引入死锁和竞态条件。多进程可以绕过GIL但进程更重且进程间通信IPC复杂。对于爬虫这种IO密集型任务异步IO才是更轻量、更优雅的解决方案。1.3 异步IO模型事件循环与协程异步爬虫的核心是事件循环Event Loop和协程Coroutine。简单理解事件循环一个永不停歇的调度器它维护一个任务队列不断检查哪些任务可以继续执行。协程可以暂停和恢复的函数。当协程执行到await时它会主动让出控制权事件循环立即切换到其他可执行的协程。在网络请求场景中协程发起请求后遇到await等待响应便让出CPU事件循环去调度其他协程发起请求。当某个请求的响应数据到达时事件循环再唤醒对应的协程继续执行。这样一来单个线程内就能实现高并发且几乎没有线程切换开销。1.4 异步爬虫的性能潜力为了让你有直观感受我们用异步方式重写上面的示例pythonimport asyncio import aiohttp import time urls [https://httpbin.org/delay/1 for _ in range(10)] async def fetch(session, url): async with session.get(url) as response: return await response.json() async def async_fetch(urls): async with aiohttp.ClientSession() as session: tasks [fetch(session, url) for url in urls] return await asyncio.gather(*tasks) start time.time() asyncio.run(async_fetch(urls)) print(f异步耗时: {time.time() - start:.2f}秒)在我本地的测试中这个异步版本耗时仅约1.2秒包括了网络往返和服务器延迟。10倍的提升就藏在这几十行代码的差异里。第二章aiohttp基础——异步HTTP客户端2.1 为什么选择aiohttp而非requestsrequests库虽然易用但它基于同步阻塞的urllib3不支持异步。虽然requests社区有requests-futures或requests-threads这类扩展但本质上还是用多线程模拟异步并非真正的异步IO。aiohttp是专门为asyncio设计的HTTP客户端也支持服务端它完全基于异步IO能够与事件循环无缝配合。此外aiohttp还提供了ClientSession可以复用TCP连接Keep-Alive进一步提升性能。2.2 安装与环境准备bashpip install aiohttp # 可选安装加速的DNS解析器 pip install aiodnsPython版本要求3.7因为使用了asyncio.run和类型注解本文代码基于Python 3.10编写。2.3 ClientSession会话管理的核心在aiohttp中所有的请求都通过ClientSession发起。它类似于requests.Session可以保存Headers、Cookies并复用连接池。使用Session的一个关键好处是复用底层的TCP连接避免每次请求都重新进行三次握手这在大量请求时能显著减少延迟。pythonimport aiohttp import asyncio async def main(): async with aiohttp.ClientSession() as session: # 所有请求都通过session发起 async with session.get(https://httpbin.org/get) as resp: print(resp.status) data await resp.json() print(data)注意async with session.get(...)是上下文管理器它会确保响应对象被正确关闭释放连接回连接池。2.4 发起GET/POST请求GET请求pythonasync with session.get(https://api.example.com/data, params{key: value}) as resp: # 文本响应 text await resp.text() # JSON响应 json_data await resp.json() # 二进制内容如图片 content await resp.read()POST请求python# 表单数据 await session.post(https://httpbin.org/post, data{key: value}) # JSON数据 await session.post(https://httpbin.org/post, json{key: value})2.5 超时设置与异常处理网络请求不可靠超时和异常处理必不可少。aiohttp提供了Timeout上下文管理器pythonimport aiohttp from aiohttp import ClientTimeout timeout ClientTimeout(total10) # 总超时10秒 async with aiohttp.ClientSession(timeouttimeout) as session: try: async with session.get(https://slow-server.com) as resp: return await resp.json() except asyncio.TimeoutError: print(请求超时跳过) except aiohttp.ClientError as e: print(f客户端错误: {e})2.6 连接池与并发控制aiohttp默认的连接池限制是100个并发连接。如果目标服务器无法承受高并发或者你的IP被限制需要手动控制并发数。可以使用asyncio.Semaphore信号量来限制同时进行的请求数量pythonsem asyncio.Semaphore(20) # 同时最多20个请求 async def fetch_with_limit(session, url): async with sem: async with session.get(url) as resp: return await resp.text()此外你还可以在创建ClientSession时调整连接池参数pythonconn aiohttp.TCPConnector(limit50, limit_per_host20) # 总连接数50单域名20 async with aiohttp.ClientSession(connectorconn) as session: ...第三章构建第一个异步爬虫3.1 目标抓取一个新闻网站的标题列表为了演示我们选择公开的测试站点如https://example.com或使用https://httpbin.org的模拟接口。但为了更真实假设我们要爬取一个博客网站的首页文章列表这里用https://httpbin.org/html代替因为它返回一个模拟的HTML页面。实际项目中你通常会针对真实的目标网站编写解析逻辑。本章我们以解析HTML中的标题为例。3.2 异步请求与HTML解析结合异步爬虫的常见模式是异步请求同步解析或者异步解析。由于解析HTML是CPU密集型操作虽然对现代CPU来说不算重用BeautifulSoup或lxml同步解析即可不会成为瓶颈。pythonimport asyncio import aiohttp from bs4 import BeautifulSoup import time async def fetch_html(session, url): try: async with session.get(url) as resp: resp.raise_for_status() return await resp.text() except Exception as e: print(f请求失败: {url}, 错误: {e}) return None async def parse_and_extract(session, url): html await fetch_html(session, url) if html is None: return None soup BeautifulSoup(html, lxml) # 示例提取所有h1标签文本真实站点可替换为具体的选择器 titles [h1.text.strip() for h1 in soup.find_all(h1)] return titles async def main(): urls [https://httpbin.org/html for _ in range(20)] # 重复20次模拟批量 async with aiohttp.ClientSession() as session: tasks [parse_and_extract(session, url) for url in urls] results await asyncio.gather(*tasks) # 统计成功数量 success [r for r in results if r is not None] print(f成功抓取 {len(success)} 个页面) start time.time() asyncio.run(main()) print(f耗时: {time.time() - start:.2f}秒)这段代码中asyncio.gather并发执行所有任务。如果某个请求失败我们捕获异常返回None避免整个程序崩溃。3.3 性能对比实验为了定量对比我写了一个同步版本使用requestsBeautifulSoup同样抓取20个页面。在我的网络环境下同步版本耗时约22秒异步版本约2.5秒——性能提升接近9倍。为什么不是理想中的20倍因为单次请求的响应时间不只是服务器延迟还包括网络RTT、DNS解析、TLS握手等。这些开销在异步模式下虽然被重叠但仍有一部分无法完全消除如TCP连接建立且httpbin.org本身有一定负载限制。但在实际抓取大型网站时提升10倍是完全可能的。第四章异步爬虫进阶——错误重试与动态并发4.1 优雅的重试机制爬虫不可避免会遇到临时故障服务器5xx错误、连接重置、DNS超时等。一个健壮的爬虫需要有自动重试机制且最好支持指数退避Exponential Backoff避免加重服务器负担。pythonimport asyncio import aiohttp from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type # 使用tenacity库简化重试逻辑 retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max5), retryretry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)) ) async def fetch_with_retry(session, url): async with session.get(url) as resp: resp.raise_for_status() return await resp.text()如果不希望引入额外库也可以手动实现pythonasync def fetch_with_manual_retry(session, url, max_retries3): for attempt in range(max_retries): try: async with session.get(url) as resp: resp.raise_for_status() return await resp.text() except Exception as e: if attempt max_retries - 1: raise wait_time 2 ** attempt # 指数退避 await asyncio.sleep(wait_time)4.2 动态调整并发数自适应流控理想情况下我们希望以最快的速度爬取但又不至于被目标网站封禁。这就需要动态调整并发数。一种简单的策略是根据失败率动态调整信号量。pythonclass AdaptiveSemaphore: def __init__(self, initial_limit20, max_limit100, min_limit5): self.limit initial_limit self.max_limit max_limit self.min_limit min_limit self.failure_count 0 self.total_count 0 self.sem asyncio.Semaphore(initial_limit) async def acquire(self): await self.sem.acquire() def release(self, successTrue): self.sem.release() self.total_count 1 if not success: self.failure_count 1 # 每10个请求检查一次调整 if self.total_count % 10 0: fail_rate self.failure_count / self.total_count if fail_rate 0.3 and self.limit self.min_limit: self.limit max(self.min_limit, self.limit - 5) self._reset_sem() elif fail_rate 0.05 and self.limit self.max_limit: self.limit min(self.max_limit, self.limit 5) self._reset_sem() def _reset_sem(self): # 重置信号量注意简单实现中可新创建但需处理正在等待的任务 self.sem asyncio.Semaphore(self.limit)不过这只是一个概念示例实际使用时需要小心处理信号量的重置导致正在等待的任务异常。更稳健的方式是使用asyncio.Queue来控制并发任务数。4.3 使用asyncio.Queue实现生产者-消费者模式对于大批量URL一次性创建所有任务会占用大量内存。更好的做法是使用队列Queue生产者不断放入URL消费者Worker从队列中取出并处理。pythonimport asyncio import aiohttp from bs4 import BeautifulSoup async def worker(session, queue, results, sem): while True: url await queue.get() if url is None: # 毒丸信号停止worker queue.task_done() break async with sem: try: async with session.get(url) as resp: text await resp.text() # 解析逻辑 soup BeautifulSoup(text, lxml) titles [h.text for h in soup.find_all(h1)] results.append((url, titles)) except Exception as e: print(fError on {url}: {e}) finally: queue.task_done() async def main(): urls [https://httpbin.org/html for _ in range(100)] queue asyncio.Queue() for url in urls: await queue.put(url) results [] sem asyncio.Semaphore(20) async with aiohttp.ClientSession() as session: workers [asyncio.create_task(worker(session, queue, results, sem)) for _ in range(10)] # 等待所有URL被处理 await queue.join() # 停止worker for _ in workers: await queue.put(None) await asyncio.gather(*workers) print(f总共处理 {len(results)} 个URL)这种模式非常灵活你可以很容易地调整worker数量来控制并发同时队列还能起到缓冲作用。第五章进阶技巧——Headers伪装、代理与Cookie持久化5.1 模拟浏览器Headers很多网站会检查User-Agent和Referer等Header。我们可以为整个Session设置默认Headerspythonheaders { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.8,en-US;q0.5,en;q0.3, } async with aiohttp.ClientSession(headersheaders) as session: ...也可以为单个请求添加额外Headerspythonasync with session.get(url, headers{X-Custom: value}) as resp: ...5.2 使用代理Proxy在爬虫被反爬时代理是常用手段。aiohttp支持HTTP/HTTPS和SOCKS代理需安装aiohttp-socks。python# HTTP代理 proxy http://user:passhost:port async with session.get(url, proxyproxy) as resp: ... # HTTPS代理 proxy https://user:passhost:port async with session.get(url, proxyproxy) as resp: ...对于大量请求轮换代理可以将代理列表传入每次随机选择一个。5.3 Cookie持久化与自动管理ClientSession内置了Cookie存储可以自动处理Set-Cookie。如果需要保存Cookie到文件可以使用http.cookiejar配合aiohttp.CookieJarpythonfrom http.cookiejar import MozillaCookieJar cookie_jar MozillaCookieJar(cookies.txt) cookie_jar.load(ignore_discardTrue) session aiohttp.ClientSession(cookie_jarcookie_jar) # 使用后保存 cookie_jar.save(ignore_discardTrue)但注意aiohttp的CookieJar与requests的略有不同更推荐在Session层面处理Cookie或者在请求间手动传递。第六章实战项目——异步爬取百万级商品数据架构设计6.1 项目需求假设我们需要从某电商网站公开数据遵守robots.txt抓取100万个商品详情页。页面结构固定我们需要提取价格、名称、评分等信息。6.2 架构分层设计一个好的爬虫项目应当分层清晰URL管理模块负责生成或读取URL列表支持断点续爬使用Redis或本地文件记录已爬取URL。请求模块封装aiohttp处理代理、重试、超时等。解析模块使用lxml或BeautifulSoup解析HTML提取数据。数据存储模块异步写入数据库如asyncpg用于PostgreSQLaiomysql用于MySQL或文件aiofiles。调度模块控制并发、限流、监控。6.3 关键代码片段异步写入文件使用aiofilespythonimport aiofiles async def save_data(data, filename): async with aiofiles.open(filename, a, encodingutf-8) as f: await f.write(json.dumps(data, ensure_asciiFalse) \n)限流使用Leaky Bucket或Token Bucket算法pythonclass RateLimiter: def __init__(self, rate): self.rate rate # 每秒请求数 self.tokens rate self.updated_at asyncio.get_event_loop().time() self.lock asyncio.Lock() async def acquire(self): async with self.lock: now asyncio.get_event_loop().time() self.tokens (now - self.updated_at) * self.rate if self.tokens self.rate: self.tokens self.rate self.updated_at now if self.tokens 1: sleep_time (1 - self.tokens) / self.rate await asyncio.sleep(sleep_time) self.tokens 0 else: self.tokens - 16.4 监控与日志异步爬虫的调试和监控必不可少。推荐使用loguru或内置logging并在关键位置记录成功/失败计数平均响应时间当前并发数剩余待爬取URL数可以用tqdm显示进度条需注意异步兼容pythonfrom tqdm.asyncio import tqdm async for _ in tqdm(range(total), desc爬取进度): await process_one()6.5 断点续爬实现最简单的断点续爬将已爬取的URL存入一个set并在启动时加载。也可以使用Redis的SET结构支持百万级数据。python# 使用Redis import aioredis redis aioredis.from_url(redis://localhost) async def mark_done(url): await redis.sadd(crawled_urls, url) async def is_done(url): return await redis.sismember(crawled_urls, url)第七章性能调优与常见坑7.1 连接池大小与并发数调优limit总连接数限制默认100。对于大规模爬取可适当调大如200-500但受限于操作系统文件描述符。limit_per_host对同一域名的连接数限制默认0无限制。建议设置避免对单个站点造成过大压力。并发数信号量一般设为连接池上限的80%左右。7.2 防止DNS缓存污染aiohttp默认使用系统DNS解析可能较慢。可安装aiodns并指定使用pythonfrom aiohttp.resolver import AsyncResolver resolver AsyncResolver(nameservers[8.8.8.8, 1.1.1.1]) conn aiohttp.TCPConnector(resolverresolver)7.3 内存泄漏防范确保每个响应对象都被正确关闭使用async with自动管理。避免在循环中积累大量未释放的对象。对于超大响应体使用resp.content.iter_chunked()流式读取而不是一次性read()。pythonasync with session.get(url) as resp: async for chunk in resp.content.iter_chunked(1024): # 处理chunk pass7.4 事件循环的优雅关闭在长时间运行的爬虫中需处理KeyboardInterrupt确保资源释放pythontry: asyncio.run(main()) except KeyboardInterrupt: print(收到中断信号正在清理...) # 取消所有任务 tasks asyncio.all_tasks() for task in tasks: task.cancel() asyncio.gather(*tasks, return_exceptionsTrue)第八章异步爬虫的局限性及替代方案8.1 什么时候不适合用异步爬虫请求量很小几百个同步爬虫足够异步优势不明显。解析逻辑极其复杂CPU密集型如果每个页面解析需要0.5秒CPU时间即使异步请求再快整体性能也会受限于CPU。此时可考虑多进程异步结合。目标网站严格限制IP并发即使你异步并发服务器也可能返回429。此时单线程加随机延迟反而更稳定。8.2 与Scrapy框架的对比Scrapy是基于异步框架Twisted构建的其性能同样优秀。Scrapy提供了完整的爬虫框架包括中间件、管道、选择器等适合大型项目。但aiohttp更轻量、更灵活适合定制化需求。如果你的项目需要快速开发Scrapy是好的选择如果追求对异步IO的完全控制aiohttp更合适。8.3 未来趋势Python 3.11对异步支持持续增强asyncio的性能也在不断优化。此外新的库如httpx也支持异步且API更接近requests可以作为aiohttp的替代。但aiohttp依然是最成熟、生态最丰富的选择。结语异步思维改变你的爬虫开发方式从同步阻塞到异步事件驱动不仅仅是换了一个库更是一种编程范式的转变。异步爬虫让我们能够以极低的资源消耗实现极高的并发这是现代爬虫工程师必须掌握的技能。本文从同步爬虫的痛点出发逐步讲解了aiohttp的基础用法、并发控制、错误处理、实战架构以及调优技巧。希望你能将这些知识应用到实际项目中让你的爬虫效率提升10倍甚至更多。最后请记住爬虫开发的黄金法则尊重网站的robots.txt控制合理的请求频率做有道德的爬虫开发者。附录完整项目代码示例可直接运行为了便于你上手这里提供一个完整的异步爬虫类封装了常用功能pythonimport asyncio import aiohttp from aiohttp import ClientTimeout, TCPConnector from typing import List, Dict, Any, Optional import logging from tenacity import retry, stop_after_attempt, wait_exponential logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class AsyncCrawler: def __init__(self, max_concurrent: int 20, max_retries: int 3, timeout: int 30, headers: Optional[Dict] None, proxy: Optional[str] None): self.max_concurrent max_concurrent self.max_retries max_retries self.timeout ClientTimeout(totaltimeout) self.headers headers or {} self.proxy proxy self.semaphore asyncio.Semaphore(max_concurrent) self.session: Optional[aiohttp.ClientSession] None async def __aenter__(self): connector TCPConnector(limit100, limit_per_host30) self.session aiohttp.ClientSession( headersself.headers, timeoutself.timeout, connectorconnector ) return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) async def fetch(self, url: str, method: str GET, **kwargs) - str: async with self.semaphore: async with self.session.request(method, url, proxyself.proxy, **kwargs) as resp: resp.raise_for_status() return await resp.text() async def fetch_json(self, url: str, **kwargs) - Dict[str, Any]: text await self.fetch(url, **kwargs) import json return json.loads(text) async def fetch_many(self, urls: List[str], parserNone) - List[Any]: tasks [] for url in urls: if parser: tasks.append(self._parse_wrapper(url, parser)) else: tasks.append(self.fetch(url)) return await asyncio.gather(*tasks, return_exceptionsTrue) async def _parse_wrapper(self, url, parser): html await self.fetch(url) return parser(html, url) # 使用示例 async def main(): urls [https://httpbin.org/html for _ in range(50)] headers {User-Agent: Mozilla/5.0} async with AsyncCrawler(max_concurrent30, headersheaders) as crawler: results await crawler.fetch_many(urls) success [r for r in results if not isinstance(r, Exception)] logger.info(f成功: {len(success)}, 失败: {len(results) - len(success)}) if __name__ __main__: asyncio.run(main())
)
