Python异步编程深度实战：从Asyncio到高性能应用开发

异步Web爬虫系统

import asyncio
import aiohttp
from urllib.parse import urljoin, urlparse
import time
from collections import deque
import json

class AsyncWebCrawler:
    def __init__(self, base_url, max_concurrent=10, max_pages=100):
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.max_pages = max_pages
        self.visited = set()
        self.to_visit = deque([base_url])
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = []
        self.session = None
    
    async def fetch_page(self, url):
        """异步获取页面内容"""
        async with self.semaphore:
            try:
                async with self.session.get(url, timeout=10) as response:
                    if response.status == 200:
                        content = await response.text()
                        return {
                            'url': url,
                            'content': content[:500],  # 只保存前500字符
                            'status': 'success',
                            'links': await self.extract_links(content, url)
                        }
                    else:
                        return {'url': url, 'status': f'error: {response.status}'}
            except Exception as e:
                return {'url': url, 'status': f'error: {str(e)}'}
    
    async def extract_links(self, html, base_url):
        """从HTML中提取链接（简化版）"""
        # 这里使用简单的字符串查找，实际项目中可使用BeautifulSoup
        import re
        links = re.findall(r'href="([^" rel="external nofollow" ]*)"', html)
        full_links = []
        
        for link in links:
            full_link = urljoin(base_url, link)
            if self.is_valid_url(full_link):
                full_links.append(full_link)
        
        return full_links
    
    def is_valid_url(self, url):
        """验证URL是否有效且属于同一域名"""
        parsed = urlparse(url)
        base_parsed = urlparse(self.base_url)
        return (parsed.netloc == base_parsed.netloc and 
                parsed.scheme in ['http', 'https'])
    
    async def crawl(self):
        """主爬取循环"""
        self.session = aiohttp.ClientSession()
        
        try:
            while (self.to_visit and 
                   len(self.visited) < self.max_pages and 
                   len(self.results) < self.max_pages):
                
                # 获取一批URL进行处理
                batch_urls = []
                while self.to_visit and len(batch_urls) < self.max_concurrent:
                    url = self.to_visit.popleft()
                    if url not in self.visited:
                        self.visited.add(url)
                        batch_urls.append(url)
                
                if not batch_urls:
                    break
                
                # 并发获取页面
                tasks = [self.fetch_page(url) for url in batch_urls]
                batch_results = await asyncio.gather(*tasks)
                
                # 处理结果并发现新链接
                for result in batch_results:
                    self.results.append(result)
                    if 'links' in result:
                        for link in result['links']:
                            if (link not in self.visited and 
                                link not in self.to_visit):
                                self.to_visit.append(link)
                
                print(f"已爬取: {len(self.results)} 页面, 待访问: {len(self.to_visit)}")
                
        finally:
            await self.session.close()
    
    def save_results(self, filename='crawl_results.json'):
        """保存爬取结果"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)
    
    async def run(self):
        """运行爬虫"""
        start_time = time.time()
        await self.crawl()
        end_time = time.time()
        
        print(f"爬取完成! 总共爬取 {len(self.results)} 个页面")
        print(f"总耗时: {end_time - start_time:.2f} 秒")
        self.save_results()

# 使用示例
async def main():
    crawler = AsyncWebCrawler(
        base_url="https://httpbin.org",  # 使用测试网站
        max_concurrent=5,
        max_pages=20
    )
    await crawler.run()

# 运行爬虫
if __name__ == "__main__":
    asyncio.run(main())