Python 异步爬虫进阶：aiohttp+asyncio 并发控制指南

一、并发控制的核心需求

1. 防止过量请求
   避免触发目标网站反爬机制（如封IP）
2. 资源优化
   控制内存/CPU消耗，避免 Too many open files 错误
3. 稳定性保障
   降低因网络波动导致的任务失败率

二、关键技术方案

信号量（Semaphore）控制
通过 $$ \text{并发量} = \frac{\text{总任务数}}{\text{信号量值}} $$ 实现精准控制

import aiohttp
import asyncio

async def fetch(url, semaphore):
    async with semaphore:  # 信号量准入控制
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()

async def main(urls, max_concurrent=10):
    semaphore = asyncio.Semaphore(max_concurrent)  # 设置并发上限
    tasks = [fetch(url, semaphore) for url in urls]
    return await asyncio.gather(*tasks)

三、进阶优化技巧

1. 动态并发调节
   根据响应时间自动调整并发密度：

   class AdaptiveSemaphore:
       def __init__(self, base=10, max=50):
           self.sem = asyncio.Semaphore(base)
           self.base = base
           self.max = max
       
       async def adjust(self, response_time):
           if response_time > 2.0 and self.sem._value > 1:
               self.sem = asyncio.Semaphore(max(1, int(self.sem._value * 0.8)))
           elif response_time < 0.5 and self.sem._value < self.max:
               self.sem = asyncio.Semaphore(min(self.max, int(self.sem._value * 1.2)))
   

2. 请求队列管理
   使用优先级队列处理重要任务：

   from heapq import heappush, heappop
   
   class PriorityQueue:
       def __init__(self):
           self._queue = []
           self._event = asyncio.Event()
       
       async def get(self):
           while not self._queue:
               await self._event.wait()
           return heappop(self._queue)[1]
       
       def put(self, priority, item):
           heappush(self._queue, (priority, item))
           self._event.set()
   

3. 错误熔断机制
   当错误率超过阈值时自动降级：

   class CircuitBreaker:
       def __init__(self, max_errors=5, cooldown=30):
           self.error_count = 0
           self.cooldown = cooldown
           self.max_errors = max_errors
           self.locked = False
       
       async def __aenter__(self):
           if self.locked:
               await asyncio.sleep(self.cooldown)
           return self
       
       def __aexit__(self, exc_type, exc, tb):
           if exc_type:
               self.error_count += 1
               if self.error_count >= self.max_errors:
                   self.locked = True
                   asyncio.create_task(self.reset_timer())
           else:
               self.error_count = max(0, self.error_count - 1)
       
       async def reset_timer(self):
           await asyncio.sleep(self.cooldown)
           self.locked = False
           self.error_count = 0
   

四、最佳实践组合方案

async def robust_crawler(urls, init_concurrent=15):
    semaphore = asyncio.Semaphore(init_concurrent)
    breaker = CircuitBreaker()
    queue = PriorityQueue()
    
    for url in urls:
        queue.put(priority=1, item=url)  # 可设置优先级
    
    async def worker():
        while True:
            url = await queue.get()
            async with breaker, semaphore:
                try:
                    # 添加超时控制
                    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session:
                        async with session.get(url) as resp:
                            data = await resp.read()
                            # 处理数据...
                except Exception as e:
                    print(f"Failed {url}: {str(e)}")
                    queue.put(priority=0, item=url)  # 失败重试
    
    workers = [asyncio.create_task(worker()) for _ in range(init_concurrent)]
    await asyncio.gather(*workers)

五、性能监控指标

建议监控以下核心指标：

1. $$ \text{请求成功率} = \frac{\text{成功数}}{\text{请求总数}} \times 100% $$
2. $$ \text{平均响应时间} = \frac{\sum{\text{单次耗时}}}{\text{请求总数}} $$
3. $$ \text{资源利用率} = \frac{\text{实际并发}}{\text{最大并发}} \times 100% $$

注意事项：

1. 初始并发值建议设置在 10-20 之间
2. 重要域名添加 aiohttp.TCPConnector(limit_per_host=5) 限制单域名连接数
3. 分布式场景需使用 Redis 信号量替代本地信号量