Python异步编程进阶：AsyncIO与结构化并发实战指南

三、分布式异步爬虫系统实战

3.1 系统架构设计

import asyncio
import aiohttp
import asyncpg
from urllib.parse import urljoin, urlparse
from bs4 import BeautifulSoup
import hashlib
from dataclasses import dataclass
from typing import Set, List, Optional
import json

@dataclass
class CrawlConfig:
    max_depth: int = 3
    max_pages: int = 1000
    politeness_delay: float = 1.0
    user_agent: str = "AsyncCrawler/1.0"
    allowed_domains: Optional[Set[str]] = None

class DistributedAsyncCrawler:
    def __init__(self, config: CrawlConfig, db_connection: str):
        self.config = config
        self.db_connection = db_connection
        self.visited_urls: Set[str] = set()
        self.session: Optional[aiohttp.ClientSession] = None
        self.pool: Optional[asyncpg.Pool] = None
        self.semaphore = asyncio.Semaphore(50)  # 控制并发数
    
    async def initialize(self):
        """初始化连接池"""
        self.session = aiohttp.ClientSession(
            headers={'User-Agent': self.config.user_agent}
        )
        self.pool = await asyncpg.create_pool(self.db_connection)
        
        # 创建数据库表
        async with self.pool.acquire() as conn:
            await conn.execute('''
                CREATE TABLE IF NOT EXISTS crawled_pages (
                    id SERIAL PRIMARY KEY,
                    url_hash VARCHAR(64) UNIQUE,
                    url TEXT NOT NULL,
                    title TEXT,
                    content TEXT,
                    depth INTEGER,
                    status_code INTEGER,
                    crawled_at TIMESTAMP DEFAULT NOW()
                )
            ''')
    
    async def close(self):
        """清理资源"""
        if self.session:
            await self.session.close()
        if self.pool:
            await self.pool.close()
    
    def should_crawl(self, url: str, current_depth: int) -> bool:
        """判断URL是否应该被爬取"""
        if current_depth > self.config.max_depth:
            return False
        
        if len(self.visited_urls) >= self.config.max_pages:
            return False
        
        parsed = urlparse(url)
        if self.config.allowed_domains and parsed.netloc not in self.config.allowed_domains:
            return False
        
        # 基本的URL过滤
        if any(ext in url.lower() for ext in ['.pdf', '.jpg', '.png', '.zip']):
            return False
        
        return True
    
    async def crawl_page(self, url: str, depth: int = 0) -> List[str]:
        """爬取单个页面并提取链接"""
        if not self.should_crawl(url, depth) or url in self.visited_urls:
            return []
        
        self.visited_urls.add(url)
        print(f"爬取深度 {depth}: {url}")
        
        async with self.semaphore:
            await asyncio.sleep(self.config.politeness_delay)
            
            try:
                async with self.session.get(url) as response:
                    if response.status != 200:
                        await self.store_page_data(url, "", depth, response.status)
                        return []
                    
                    html = await response.text()
                    soup = BeautifulSoup(html, 'html.parser')
                    
                    # 提取页面数据
                    title = soup.title.string if soup.title else ""
                    content = soup.get_text()[:5000]  # 限制内容长度
                    
                    await self.store_page_data(url, content, depth, response.status, title)
                    
                    # 提取并返回新链接
                    new_urls = []
                    for link in soup.find_all('a', href=True):
                        absolute_url = urljoin(url, link['href'])
                        if self.should_crawl(absolute_url, depth + 1):
                            new_urls.append(absolute_url)
                    
                    return new_urls
            
            except Exception as e:
                print(f"爬取失败 {url}: {e}")
                await self.store_page_data(url, "", depth, 0, error=str(e))
                return []
    
    async def store_page_data(self, url: str, content: str, depth: int, 
                            status_code: int, title: str = "", error: str = ""):
        """存储爬取的数据到数据库"""
        url_hash = hashlib.sha256(url.encode()).hexdigest()
        
        async with self.pool.acquire() as conn:
            await conn.execute('''
                INSERT INTO crawled_pages 
                (url_hash, url, title, content, depth, status_code)
                VALUES ($1, $2, $3, $4, $5, $6)
                ON CONFLICT (url_hash) DO NOTHING
            ''', url_hash, url, title, content, depth, status_code)
    
    async def start_crawling(self, seed_urls: List[str]):
        """启动分布式爬取"""
        await self.initialize()
        
        try:
            current_level = seed_urls
            depth = 0
            
            while current_level and depth = self.config.max_pages:
                    print("达到最大页面限制，停止爬取")
                    break
        
        finally:
            await self.close()
    
    async def get_crawl_statistics(self) -> dict:
        """获取爬取统计信息"""
        async with self.pool.acquire() as conn:
            total_pages = await conn.fetchval("SELECT COUNT(*) FROM crawled_pages")
            by_depth = await conn.fetch(
                "SELECT depth, COUNT(*) FROM crawled_pages GROUP BY depth ORDER BY depth"
            )
            by_status = await conn.fetch(
                "SELECT status_code, COUNT(*) FROM crawled_pages GROUP BY status_code"
            )
            
            return {
                "total_pages": total_pages,
                "by_depth": dict(by_depth),
                "by_status": dict(by_status),
                "visited_urls": len(self.visited_urls)
            }

3.2 监控与容错机制

class CrawlerMonitor:
    def __init__(self, crawler: DistributedAsyncCrawler):
        self.crawler = crawler
        self.metrics = {
            'pages_crawled': 0,
            'errors': 0,
            'start_time': None,
            'last_checkpoint': None
        }
    
    async def start_monitoring(self):
        """启动监控循环"""
        self.metrics['start_time'] = asyncio.get_event_loop().time()
        self.metrics['last_checkpoint'] = self.metrics['start_time']
        
        while True:
            await asyncio.sleep(5)  # 每5秒报告一次
            await self.report_status()
    
    async def report_status(self):
        """报告当前状态"""
        current_time = asyncio.get_event_loop().time()
        elapsed = current_time - self.metrics['start_time']
        since_last = current_time - self.metrics['last_checkpoint']
        
        stats = await self.crawler.get_crawl_statistics()
        pages_per_second = stats['visited_urls'] / elapsed if elapsed > 0 else 0
        
        print(f"n=== 爬虫状态报告 ===")
        print(f"运行时间: {elapsed:.1f}秒")
        print(f"已爬取页面: {stats['visited_urls']}")
        print(f"爬取速率: {pages_per_second:.2f} 页面/秒")
        print(f"各深度分布: {stats['by_depth']}")
        print(f"状态码分布: {stats['by_status']}")
        
        self.metrics['last_checkpoint'] = current_time

# 使用示例
async def main():
    config = CrawlConfig(
        max_depth=2,
        max_pages=100,
        allowed_domains={'example.com', 'test.org'}
    )
    
    crawler = DistributedAsyncCrawler(
        config=config,
        db_connection="postgresql://user:pass@localhost/crawler"
    )
    
    monitor = CrawlerMonitor(crawler)
    
    # 启动监控任务
    monitor_task = asyncio.create_task(monitor.start_monitoring())
    
    # 开始爬取
    seed_urls = [
        'https://example.com',
        'https://example.com/page1',
        'https://example.com/page2'
    ]
    
    try:
        await crawler.start_crawling(seed_urls)
    finally:
        monitor_task.cancel()
        try:
            await monitor_task
        except asyncio.CancelledError:
            pass
        
        # 最终统计报告
        final_stats = await crawler.get_crawl_statistics()
        print(f"n=== 最终统计 ===")
        print(json.dumps(final_stats, indent=2))

if __name__ == "__main__":
    asyncio.run(main())