Python中大量爬虫如何优化？服务器带宽有限时同时爬取几万个网站的解决方案

我这里举个例子，比如要跑一万个网站，单机，带宽 4M，异步跑的话，必然会有很多网站传回响应会在带宽方面被限制，最次解决办法：是否增加网站的超时时间可以缓解，高级一点：可以通过引入队列，判断任务并发个数，来判断是否执行任务，来增加网站的传回响应时间以及减少网站丢失数据的可能性，更高级一点：你们来说！！！！

sinazl 1楼

再跑一台 10M 带宽的机器, 只用来下载, 下载数据直接通过 redis 之类的工具传给解析器
这样做的好处:
1. 把瓶颈分离: 下载服务器只需要关注带宽, 解析服务器只需要关注 CPU 和数据库, 配置不足升级也方便
2. 好扩展: 你可以随时添加任意数量的下载服务器

sinazl 2楼

这个问题我遇到过，核心是控制并发数+优化请求策略。服务器带宽有限就别硬刚了，得用巧劲。

直接上代码，我用aiohttp+asyncio做异步爬虫，配合信号量控制并发：

import aiohttp
import asyncio
from asyncio import Semaphore
import time
from typing import List, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class OptimizedCrawler:
    def __init__(self, max_concurrent: int = 50, timeout: int = 30):
        self.max_concurrent = max_concurrent
        self.timeout = timeout
        self.semaphore = Semaphore(max_concurrent)
        self.session = None
        
    async def fetch_single(self, session: aiohttp.ClientSession, url: str) -> Dict:
        """单个URL的爬取逻辑"""
        async with self.semaphore:  # 信号量控制并发
            try:
                async with session.get(url, timeout=self.timeout) as response:
                    content = await response.text()
                    return {
                        'url': url,
                        'status': response.status,
                        'content_length': len(content),
                        'success': True
                    }
            except Exception as e:
                logger.warning(f"Failed to fetch {url}: {str(e)}")
                return {'url': url, 'success': False, 'error': str(e)}
    
    async def crawl_batch(self, urls: List[str]) -> List[Dict]:
        """批量爬取"""
        connector = aiohttp.TCPConnector(limit=self.max_concurrent)  # 连接器限制
        timeout = aiohttp.ClientTimeout(total=self.timeout)
        
        async with aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={'User-Agent': 'Mozilla/5.0'}
        ) as session:
            
            tasks = [self.fetch_single(session, url) for url in urls]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 过滤异常结果
            valid_results = []
            for result in results:
                if not isinstance(result, Exception):
                    valid_results.append(result)
            
            return valid_results
    
    def run(self, urls: List[str]) -> List[Dict]:
        """同步入口"""
        return asyncio.run(self.crawl_batch(urls))

# 使用示例
if __name__ == "__main__":
    # 模拟几万个URL（实际使用时替换为真实URL列表）
    urls = [f"https://httpbin.org/get?page={i}" for i in range(100)]
    
    crawler = OptimizedCrawler(max_concurrent=30)  # 根据带宽调整并发数
    
    start_time = time.time()
    results = crawler.run(urls)
    elapsed = time.time() - start_time
    
    success_count = sum(1 for r in results if r.get('success'))
    print(f"爬取完成: {success_count}/{len(urls)} 成功, 耗时: {elapsed:.2f}秒")

关键优化点：