python中aiohttp异步高并发爬虫实战代码指南

semaphore = asyncio.Semaphore(100)  # 限制最大并发100
 
async def fetch_url(session, url):
    async with semaphore:  # 获取信号量许可
        try:
            async with session.get(url, timeout=10) as response:
                return await response.text()
        except Exception as e:
            return f"Error: {str(e)}"

connector = aiohttp.TCPConnector(limit=0)  # 0表示不限制连接数
async with aiohttp.ClientSession(connector=connector) as session:
    # 复用TCP连接处理多个请求
    pass

async def robust_fetch(session, url):
    for _ in range(3):  # 自动重试3次
        try:
            async with session.get(url, timeout=10) as response:
                if response.status == 200:
                    return await response.text()
                elif response.status == 429:  # 触发反爬
                    await asyncio.sleep(5)  # 指数退避
                    continue
        except (aiohttp.ClientError, asyncio.TimeoutError):
            await asyncio.sleep(1)  # 短暂等待后重试
    return f"Failed: {url}"

import asyncio
import aiohttp
from datetime import datetime
 
async def fetch(session, url):
    start_time = datetime.now()
    try:
        async with session.get(url, timeout=10) as response:
            content = await response.text()
            return {
                "url": url,
                "status": response.status,
                "length": len(content),
                "time": (datetime.now() - start_time).total_seconds()
            }
    except Exception as e:
        return {"url": url, "error": str(e)}
 
async def crawl(urls, max_concurrency=50):
    semaphore = asyncio.Semaphore(max_concurrency)
    connector = aiohttp.TCPConnector(limit=0)
    
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results
 
if __name__ == "__main__":
    test_urls = ["https://httpbin.org/get?q={i}" for i in range(30)]
    start = datetime.now()
    results = asyncio.run(crawl(test_urls))
    elapsed = (datetime.now() - start).total_seconds()
    
    success = [r for r in results if "error" not in r]
    print(f"完成! 耗时: {elapsed:.2f}秒")
    print(f"成功率: {len(success)/len(results):.1%}")

import asyncio
import aiohttp
import hashlib
from pathlib import Path
 
class AdvancedCrawler:
    def __init__(self, max_concurrency=100, retry_times=3):
        self.max_concurrency = max_concurrency
        self.retry_times = retry_times
        self.semaphore = None
        self.session = None
        
    async def initialize(self):
        self.semaphore = asyncio.Semaphore(self.max_concurrency)
        connector = aiohttp.TCPConnector(limit=0)
        self.session = aiohttp.ClientSession(
            connector=connector,
            headers={"User-Agent": "Mozilla/5.0"},
            timeout=aiohttp.ClientTimeout(total=15)
        )
    
    async def fetch_with_retry(self, url):
        for attempt in range(self.retry_times):
            try:
                async with self.semaphore:
                    async with self.session.get(url) as response:
                        if response.status == 200:
                            return await self._save_content(url, await response.text())
                        elif response.status == 429:
                            await asyncio.sleep(2 ** attempt)  # 指数退避
                            continue
            except (aiohttp.ClientError, asyncio.TimeoutError):
                if attempt == self.retry_times - 1:
                    return f"Failed after {self.retry_times} attempts: {url}"
                await asyncio.sleep(1)
    
    async def _save_content(self, url, content):
        url_hash = hashlib.md5(url.encode()).hexdigest()
        Path("data").mkdir(exist_ok=True)
        with open(f"data/{url_hash}.html", "w", encoding="utf-8") as f:
            f.write(content)
        return {"url": url, "status": "saved"}
    
    async def close(self):
        await self.session.close()
 
# 使用示例
async def main():
    crawler = AdvancedCrawler(max_concurrency=200)
    await crawler.initialize()
    
    urls = [f"https://example.com/page/{i}" for i in range(1000)]
    tasks = [crawler.fetch_with_retry(url) for url in urls]
    await asyncio.gather(*tasks)
    await crawler.close()
 
asyncio.run(main())

async def fetch_with_proxy(session, url, proxy_url):
    try:
        async with session.get(
            url,
            proxy=proxy_url,
            proxy_auth=aiohttp.BasicAuth("user", "pass")  # 如果需要认证
        ) as response:
            return await response.text()
    except Exception as e:
        return f"Proxy Error: {str(e)}"
 
# 使用示例
proxies = [
    "http://proxy1.example.com:8080",
    "http://proxy2.example.com:8080"
]
 
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_proxy(session, "https://target.com", proxy)
            for proxy in proxies
        ]
        results = await asyncio.gather(*tasks)

async def crawl_dynamic_urls(base_url, start_page, end_page):
    semaphore = asyncio.Semaphore(100)
    
    async def fetch_page(page_num):
        url = f"{base_url}?page={page_num}"
        async with semaphore:
            async with aiohttp.ClientSession().get(url) as resp:
                return await resp.text()
    
    tasks = [fetch_page(i) for i in range(start_page, end_page + 1)]
    return await asyncio.gather(*tasks)
 
# 抓取第1-100页
results = asyncio.run(crawl_dynamic_urls("https://example.com/news", 1, 100))

import redis
import asyncio
import aiohttp
 
async def worker():
    r = redis.Redis(host='master-ip', port=6379)
    semaphore = asyncio.Semaphore(50)
    
    async with aiohttp.ClientSession() as session:
        while True:
            url = await r.blpop("url_queue")  # 阻塞式获取任务
            if not url:
                break
                
            async with semaphore:
                try:
                    async with session.get(url[1].decode()) as resp:
                        content = await resp.text()
                        await r.rpush("result_queue", content)
                except Exception as e:
                    await r.rpush("error_queue", f"{url[1]}: {str(e)}")
 
asyncio.run(worker())

# 随机User-Agent生成
import random
from fake_useragent import UserAgent
 
ua = UserAgent()
headers = {
    "User-Agent": ua.random,
    "Accept-Language": "en-US,en;q=0.9",
    "Referer": "https://www.google.com/"
}
 
# 请求间隔随机化
async def fetch_with_jitter(session, url):
    delay = random.uniform(0.5, 3.0)  # 0.5-3秒随机延迟
    await asyncio.sleep(delay)
    async with session.get(url) as resp:
        return await resp.text()

import logging
 
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("crawler.log"),
        logging.StreamHandler()
    ]
)
 
logger = logging.getLogger(__name__)
 
async def fetch(session, url):
    logger.info(f"Starting request to {url}")
    try:
        async with session.get(url) as resp:
            logger.info(f"Success: {url} - {resp.status}")
            return await resp.text()
    except Exception as e:
        logger.error(f"Failed {url}: {str(e)}")

FROM python:3.9-slim
 
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt --no-cache-dir
 
COPY . .
CMD ["python", "crawler.py"]

# 增加重试逻辑和更短的超时设置
async with session.get(
    url,
    timeout=aiohttp.ClientTimeout(total=5, connect=2)  # 更短的连接超时
) as resp:
    pass

async def safe_fetch():
    session = aiohttp.ClientSession()
    try:
        async with session.get(url) as resp:
            return await resp.text()
    finally:
        await session.close()  # 确保关闭会话

# 使用自定义DNS解析器
import aiodns
 
resolver = aiodns.DNSResolver()
connector = aiohttp.TCPConnector(
    resolver=resolver,
    family=socket.AF_INET  # 强制使用IPv4
)