Python异步爬虫实战：aiohttp并发采集与验证码异步处理完整教程
前言
爬虫效率是每个数据工程师都关心的问题。当你需要采集上万个页面时，同步请求一个一个排队等待的方式实在太慢了。
Python的asyncio + aiohttp组合可以让你的爬虫速度提升10-50倍，而且代码改动并不大。
本文将从零开始讲解异步爬虫的原理和实战，包括并发控制、错误处理、以及如何在异步流程中处理验证码。
同步 vs 异步：为什么差这么多
同步爬虫的瓶颈
importrequestsimporttime urls=[f'https://httpbin.org/delay/1'for_inrange(10)]start=time.time()forurlinurls:resp=requests.get(url)print(f'{resp.status_code}',end=' ')print(f'\n同步耗时:{time.time()-start:.1f}秒')# 输出: 同步耗时: 10.3秒 (每个请求等1秒，串行排队)问题很明显：每个请求都在等网络IO返回，CPU其实在空闲。
异步爬虫的优势
importaiohttpimportasyncioimporttimeasyncdeffetch(session,url):asyncwithsession.get(url)asresp:returnresp.statusasyncdefmain():urls=[f'https://httpbin.org/delay/1'for_inrange(10)]asyncwithaiohttp.ClientSession()assession:tasks=[fetch(session,url)forurlinurls]results=awaitasyncio.gather(*tasks)print(f'状态码:{results}')start=time.time()asyncio.run(main())print(f'异步耗时:{time.time()-start:.1f}秒')# 输出: 异步耗时: 1.2秒 (10个请求并发，几乎同时完成)10个请求从10秒变成1秒，这就是异步的威力。
aiohttp基础
安装
pipinstallaiohttpSession管理
importaiohttpimportasyncioasyncdefmain():# 创建session（复用TCP连接，性能更好）asyncwithaiohttp.ClientSession()assession:# GET请求asyncwithsession.get('https://httpbin.org/get')asresp:data=awaitresp.json()print(data)# POST请求asyncwithsession.post('https://httpbin.org/post',json={'key':'value'})asresp:data=awaitresp.json()print(data)# 自定义请求头headers={'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0','Accept-Language':'zh-CN,zh;q=0.9,en;q=0.8',}asyncwithsession.get('https://httpbin.org/headers',headers=headers)asresp:print(awaitresp.json())asyncio.run(main())超时和代理
importaiohttpimportasyncioasyncdefmain():# 设置超时timeout=aiohttp.ClientTimeout(total=30,connect=10)# 设置代理proxy='http://user:pass@proxy:8080'asyncwithaiohttp.ClientSession(timeout=timeout)assession:asyncwithsession.get('https://httpbin.org/ip',proxy=proxy)asresp:print(awaitresp.json())asyncio.run(main())并发控制：Semaphore
不加限制地并发会导致目标服务器拒绝连接，甚至把自己的IP封了。用Semaphore控制并发数：
importaiohttpimportasyncioasyncdeffetch_with_limit(sem,session,url):asyncwithsem:# 信号量控制并发try:asyncwithsession.get(url)asresp:text=awaitresp.text()return{'url':url,'status':resp.status,'length':len(text)}exceptExceptionase:return{'url':url,'error':str(e)}asyncdefmain():urls=[f'https://httpbin.org/get?page={i}'foriinrange(100)]sem=asyncio.Semaphore(10)# 最多同时10个请求connector=aiohttp.TCPConnector(limit=20)# TCP连接池上限asyncwithaiohttp.ClientSession(connector=connector)assession:tasks=[fetch_with_limit(sem,session,url)forurlinurls]results=awaitasyncio.gather(*tasks)success=[rforrinresultsif'error'notinr]failed=[rforrinresultsif'error'inr]print(f'成功:{len(success)}, 失败:{len(failed)}')asyncio.run(main())实战：异步爬虫完整模板
importaiohttpimportasynciofromdataclassesimportdataclassfromtypingimportOptionalimportloggingimportrandom logging.basicConfig(level=logging.INFO)logger=logging.getLogger(__name__)@dataclassclassScrapedItem:url:strtitle:strcontent:strstatus:intclassAsyncScraper:def__init__(self,concurrency=10,delay_range=(0.5,2.0),proxy=None):self.concurrency=concurrency self.delay_range=delay_range self.proxy=proxy self.sem=asyncio.Semaphore(concurrency)self.results=[]self.errors=[]asyncdeffetch_page(self,session,url):asyncwithself.sem:# 随机延迟，避免请求过于规律awaitasyncio.sleep(random.uniform(*self.delay_range))try:asyncwithsession.get(url,proxy=self.proxy)asresp:ifresp.status==200:html=awaitresp.text()item=self.parse(url,html,resp.status)self.results.append(item)logger.info(f'[OK]{url}')returnitemelifresp.status==403:logger.warning(f'[403]{url}- 可能需要处理验证码')self.errors.append({'url':url,'status':403})else:logger.warning(f'[{resp.status}]{url}')self.errors.append({'url':url,'status':resp.status})exceptasyncio.TimeoutError:logger.error(f'[TIMEOUT]{url}')self.errors.append({'url':url,'error':'timeout'})exceptExceptionase:logger.error(f'[ERROR]{url}:{e}')self.errors.append({'url':url,'error':str(e)})defparse(self,url,html,status):# 替换为你的解析逻辑fromhtml.parserimportHTMLParser title=''classTitleParser(HTMLParser):defhandle_starttag(self,tag,attrs):nonlocaltitle self._in_title=tag=='title'defhandle_data(self,data):nonlocaltitleifgetattr(self,'_in_title',False):title=data self._in_title=Falseparser=TitleParser()parser.feed(html[:5000])returnScrapedItem(url=url,title=title,content=html[:500],status=status)asyncdefrun(self,urls):timeout=aiohttp.ClientTimeout(total=30)connector=aiohttp.TCPConnector(limit=self.concurrency*2)headers={'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/124.0.0.0','Accept':'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8','Accept-Language':'zh-CN,zh;q=0.9,en;q=0.8',}asyncwithaiohttp.ClientSession(timeout=timeout,connector=connector,headers=headers)assession:tasks=[self.fetch_page(session,url)forurlinurls]awaitasyncio.gather(*tasks,return_exceptions=True)logger.info(f'完成:{len(self.results)}成功,{len(self.errors)}失败')returnself.results# 使用asyncdefmain():scraper=AsyncScraper(concurrency=5,delay_range=(1.0,3.0))urls=[f'https://httpbin.org/get?id={i}'foriinrange(50)]results=awaitscraper.run(urls)foriteminresults[:3]:print(f'{item.title}-{item.url}')asyncio.run(main())异步流程中处理验证码
异步爬虫遇到验证码时，不能阻塞整个事件循环。正确做法是将验证码解决也异步化：
importaiohttpimportasynciofrompassxapiimportAsyncPassXAPIclassAsyncScraperWithCaptcha:def__init__(self,captcha_api_key,concurrency=10):self.sem=asyncio.Semaphore(concurrency)self.solver=AsyncPassXAPI(api_key=captcha_api_key)asyncdeffetch_with_captcha(self,session,url):asyncwithself.sem:asyncwithsession.get(url)asresp:html=awaitresp.text()# 检测验证码if'data-sitekey'inhtml:token=awaitself._solve_captcha(html,url)iftoken:asyncwithsession.post(url,data={'cf-turnstile-response':token,'g-recaptcha-response':token,})asretry_resp:returnawaitretry_resp.text()returnhtmlasyncdef_solve_captcha(self,html,url):importrematch=re.search(r'data-sitekey="([^"]+)"',html)ifnotmatch:returnNonesitekey=match.group(1)if'cf-turnstile'inhtml:result=awaitself.solver.solve_turnstile(sitekey=sitekey,url=url)elif'h-captcha'inhtml:result=awaitself.solver.solve_hcaptcha(sitekey=sitekey,url=url)else:result=awaitself.solver.solve_recaptcha(sitekey=sitekey,url=url)returnresult.get('token')asyncdefrun(self,urls):asyncwithaiohttp.ClientSession()assession:tasks=[self.fetch_with_captcha(session,url)forurlinurls]returnawaitasyncio.gather(*tasks,return_exceptions=True)# 使用asyncdefmain():scraper=AsyncScraperWithCaptcha(captcha_api_key='your_passxapi_key',concurrency=10)urls=['https://protected-site.com/page/1','https://protected-site.com/page/2']results=awaitscraper.run(urls)asyncio.run(main())生产者-消费者模式
对于大规模爬取，推荐使用asyncio.Queue实现生产者-消费者模式：
importasyncioimportaiohttpasyncdefproducer(queue,urls):forurlinurls:awaitqueue.put(url)# 发送结束信号for_inrange(5):# worker数量awaitqueue.put(None)asyncdefconsumer(queue,session,results,worker_id):whileTrue:url=awaitqueue.get()ifurlisNone:breaktry:asyncwithsession.get(url)asresp:data=awaitresp.text()results.append({'url':url,'length':len(data)})print(f'[Worker-{worker_id}]{url}->{len(data)}bytes')exceptExceptionase:print(f'[Worker-{worker_id}] Error:{url}->{e}')queue.task_done()asyncdefmain():urls=[f'https://httpbin.org/get?id={i}'foriinrange(30)]queue=asyncio.Queue(maxsize=20)results=[]asyncwithaiohttp.ClientSession()assession:# 启动1个生产者 + 5个消费者producer_task=asyncio.create_task(producer(queue,urls))workers=[asyncio.create_task(consumer(queue,session,results,i))foriinrange(5)]awaitproducer_taskawaitasyncio.gather(*workers)print(f'总计采集:{len(results)}个页面')asyncio.run(main())性能优化技巧
1. 连接池复用
# 配置TCP连接池connector=aiohttp.TCPConnector(limit=100,# 总连接数上限limit_per_host=10,# 单个域名连接数ttl_dns_cache=300,# DNS缓存5分钟enable_cleanup_closed=True,)```### 2. 流式读取大文件```pythonasyncdefdownload_file(session,url,filepath):asyncwithsession.get(url)asresp:withopen(filepath,'wb')asf:asyncforchunkinresp.content.iter_chunked(8192):f.write(chunk)```### 3. 优雅关闭```pythonimportsignalasyncdefgraceful_shutdown(scraper):print('收到退出信号，正在优雅关闭...')scraper.running=False# 等待当前任务完成awaitasyncio.sleep(2)loop=asyncio.get_event_loop()loop.add_signal_handler(signal.SIGINT,lambda:asyncio.create_task(graceful_shutdown(scraper)))常见坑
- 不要在async函数中使用time.sleep：会阻塞整个事件循环，用
await asyncio.sleep() - Session要复用：每个请求创建新Session浪费TCP连接
- 并发不是越多越好：过高并发会触发反爬，建议5-20
- 异常要捕获：一个任务的异常不应该影响其他任务
总结
异步爬虫是提升采集效率的最有效手段：
- aiohttp + asyncio可以轻松实现10倍以上的速度提升
- 用Semaphore控制并发数，避免被封IP
- 验证码解决也要异步化，不能阻塞事件循环
- 生产者-消费者模式适合大规模采集场景
验证码异步解决方案可以参考：passxapi-python，提供AsyncPassXAPI异步客户端，完美融入asyncio工作流。
- 生产者-消费者模式适合大规模采集场景
觉得有帮助请点赞收藏，有问题欢迎评论区讨论。