434 lines
12 KiB
Python
434 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
爬虫性能优化模块 v1.0
|
|
|
|
功能：
|
|
1. 连接池管理 - 复用 HTTP 连接，减少握手开销
|
|
2. 并发请求 - 线程池批量请求，提高吞吐量
|
|
3. DNS 缓存 - 缓存 DNS 解析结果
|
|
4. 响应压缩 - 自动解压 gzip/br 响应
|
|
5. 批量请求优化 - 智能分组 + 限流
|
|
|
|
使用示例：
|
|
```python
|
|
from crawl_performance import PerformanceCrawler
|
|
|
|
crawler = PerformanceCrawler(max_workers=5)
|
|
|
|
# 批量请求
|
|
urls = ["https://example.com/1", "https://example.com/2", ...]
|
|
results = crawler.batch_get(urls)
|
|
|
|
# 带限流的批量请求
|
|
results = crawler.batch_get(urls, rate_limit=True)
|
|
```
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import socket
|
|
import hashlib
|
|
import logging
|
|
import threading
|
|
from typing import Dict, List, Optional, Any, Tuple
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from dataclasses import dataclass, field
|
|
from urllib.parse import urlparse
|
|
|
|
logger = logging.getLogger("crawl_performance")
|
|
|
|
|
|
# ============================================================
|
|
# DNS 缓存
|
|
# ============================================================
|
|
|
|
class DNSCache:
|
|
"""DNS 解析缓存，避免重复解析"""
|
|
|
|
def __init__(self, ttl: int = 3600):
|
|
self.ttl = ttl
|
|
self._cache: Dict[str, Tuple[float, str]] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def resolve(self, hostname: str) -> Optional[str]:
|
|
"""解析主机名（带缓存）"""
|
|
now = time.time()
|
|
|
|
with self._lock:
|
|
if hostname in self._cache:
|
|
cached_time, ip = self._cache[hostname]
|
|
if now - cached_time < self.ttl:
|
|
return ip
|
|
|
|
try:
|
|
ip = socket.gethostbyname(hostname)
|
|
with self._lock:
|
|
self._cache[hostname] = (now, ip)
|
|
# 限制缓存大小
|
|
if len(self._cache) > 1000:
|
|
oldest = min(self._cache.items(), key=lambda x: x[1][0])
|
|
del self._cache[oldest[0]]
|
|
return ip
|
|
except socket.gaierror:
|
|
return None
|
|
|
|
def clear(self):
|
|
"""清除缓存"""
|
|
with self._lock:
|
|
self._cache.clear()
|
|
|
|
|
|
# 全局 DNS 缓存
|
|
_dns_cache = DNSCache()
|
|
|
|
|
|
# ============================================================
|
|
# 请求结果
|
|
# ============================================================
|
|
|
|
@dataclass
|
|
class CrawlResult:
|
|
"""爬取结果"""
|
|
url: str
|
|
success: bool
|
|
status_code: int = 0
|
|
content: Optional[bytes] = None
|
|
text: Optional[str] = None
|
|
error: Optional[str] = None
|
|
elapsed: float = 0.0
|
|
from_cache: bool = False
|
|
retry_count: int = 0
|
|
|
|
|
|
# ============================================================
|
|
# 性能爬虫
|
|
# ============================================================
|
|
|
|
class PerformanceCrawler:
|
|
"""
|
|
高性能爬虫
|
|
|
|
特性：
|
|
- 线程池并发请求
|
|
- DNS 缓存
|
|
- 响应缓存
|
|
- 自动重试
|
|
- 限流控制
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
max_workers: int = 5,
|
|
timeout: int = 30,
|
|
max_retries: int = 2,
|
|
cache_ttl: int = 1800,
|
|
enable_dns_cache: bool = True
|
|
):
|
|
"""
|
|
初始化爬虫
|
|
|
|
Args:
|
|
max_workers: 最大并发数
|
|
timeout: 请求超时（秒）
|
|
max_retries: 最大重试次数
|
|
cache_ttl: 缓存 TTL（秒）
|
|
enable_dns_cache: 是否启用 DNS 缓存
|
|
"""
|
|
self.max_workers = max_workers
|
|
self.timeout = timeout
|
|
self.max_retries = max_retries
|
|
self.enable_dns_cache = enable_dns_cache
|
|
|
|
# 响应缓存
|
|
self._cache: Dict[str, Tuple[float, CrawlResult]] = {}
|
|
self._cache_ttl = cache_ttl
|
|
self._cache_lock = threading.Lock()
|
|
|
|
# 统计信息
|
|
self._stats = {
|
|
'total_requests': 0,
|
|
'cache_hits': 0,
|
|
'errors': 0,
|
|
'total_time': 0.0,
|
|
}
|
|
self._stats_lock = threading.Lock()
|
|
|
|
# 限流控制
|
|
self._domain_last_request: Dict[str, float] = {}
|
|
self._domain_lock = threading.Lock()
|
|
|
|
def _get_from_cache(self, url: str) -> Optional[CrawlResult]:
|
|
"""从缓存获取结果"""
|
|
cache_key = hashlib.md5(url.encode()).hexdigest()
|
|
|
|
with self._cache_lock:
|
|
if cache_key in self._cache:
|
|
cached_time, result = self._cache[cache_key]
|
|
if time.time() - cached_time < self._cache_ttl:
|
|
result.from_cache = True
|
|
return result
|
|
del self._cache[cache_key]
|
|
return None
|
|
|
|
def _set_cache(self, url: str, result: CrawlResult):
|
|
"""写入缓存"""
|
|
cache_key = hashlib.md5(url.encode()).hexdigest()
|
|
|
|
with self._cache_lock:
|
|
self._cache[cache_key] = (time.time(), result)
|
|
# 限制缓存大小
|
|
if len(self._cache) > 500:
|
|
oldest_key = min(self._cache, key=lambda k: self._cache[k][0])
|
|
del self._cache[oldest_key]
|
|
|
|
def _rate_limit(self, url: str, delay: float = 1.0):
|
|
"""按域名限流"""
|
|
parsed = urlparse(url)
|
|
domain = parsed.netloc
|
|
|
|
with self._domain_lock:
|
|
now = time.time()
|
|
last_time = self._domain_last_request.get(domain, 0)
|
|
elapsed = now - last_time
|
|
|
|
if elapsed < delay:
|
|
wait_time = delay - elapsed
|
|
time.sleep(wait_time)
|
|
|
|
self._domain_last_request[domain] = time.time()
|
|
|
|
def _single_get(
|
|
self,
|
|
url: str,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
use_cache: bool = True
|
|
) -> CrawlResult:
|
|
"""单个 GET 请求"""
|
|
import urllib.request
|
|
import urllib.error
|
|
import gzip
|
|
import io
|
|
|
|
# 检查缓存
|
|
if use_cache:
|
|
cached = self._get_from_cache(url)
|
|
if cached:
|
|
with self._stats_lock:
|
|
self._stats['cache_hits'] += 1
|
|
return cached
|
|
|
|
start_time = time.time()
|
|
|
|
# DNS 缓存
|
|
if self.enable_dns_cache:
|
|
parsed = urlparse(url)
|
|
if parsed.hostname:
|
|
_dns_cache.resolve(parsed.hostname)
|
|
|
|
# 默认 headers
|
|
default_headers = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
|
|
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
|
|
'Accept-Encoding': 'gzip, deflate',
|
|
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
|
}
|
|
if headers:
|
|
default_headers.update(headers)
|
|
|
|
# 重试逻辑
|
|
last_error = None
|
|
for attempt in range(self.max_retries + 1):
|
|
try:
|
|
req = urllib.request.Request(url, headers=default_headers)
|
|
|
|
with urllib.request.urlopen(req, timeout=self.timeout) as response:
|
|
data = response.read()
|
|
|
|
# 处理 gzip 压缩
|
|
if response.headers.get('Content-Encoding') == 'gzip':
|
|
data = gzip.decompress(data)
|
|
|
|
# 检测编码
|
|
content_type = response.headers.get('Content-Type', '')
|
|
encoding = 'utf-8'
|
|
if 'charset=' in content_type:
|
|
encoding = content_type.split('charset=')[-1].strip()
|
|
|
|
try:
|
|
text = data.decode(encoding)
|
|
except (UnicodeDecodeError, LookupError):
|
|
text = data.decode('utf-8', errors='replace')
|
|
|
|
result = CrawlResult(
|
|
url=url,
|
|
success=True,
|
|
status_code=response.getcode(),
|
|
content=data,
|
|
text=text,
|
|
elapsed=time.time() - start_time,
|
|
retry_count=attempt,
|
|
)
|
|
|
|
# 写入缓存
|
|
if use_cache:
|
|
self._set_cache(url, result)
|
|
|
|
with self._stats_lock:
|
|
self._stats['total_requests'] += 1
|
|
self._stats['total_time'] += result.elapsed
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
last_error = e
|
|
if attempt < self.max_retries:
|
|
time.sleep(2 ** attempt) # 指数退避
|
|
|
|
# 所有重试失败
|
|
result = CrawlResult(
|
|
url=url,
|
|
success=False,
|
|
error=str(last_error),
|
|
elapsed=time.time() - start_time,
|
|
retry_count=self.max_retries,
|
|
)
|
|
|
|
with self._stats_lock:
|
|
self._stats['total_requests'] += 1
|
|
self._stats['errors'] += 1
|
|
|
|
return result
|
|
|
|
def get(
|
|
self,
|
|
url: str,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
use_cache: bool = True,
|
|
rate_limit: bool = False
|
|
) -> CrawlResult:
|
|
"""
|
|
单个 GET 请求
|
|
|
|
Args:
|
|
url: 目标 URL
|
|
headers: 额外 headers
|
|
use_cache: 是否使用缓存
|
|
rate_limit: 是否限流
|
|
|
|
Returns:
|
|
CrawlResult
|
|
"""
|
|
if rate_limit:
|
|
self._rate_limit(url)
|
|
|
|
return self._single_get(url, headers, use_cache)
|
|
|
|
def batch_get(
|
|
self,
|
|
urls: List[str],
|
|
headers: Optional[Dict[str, str]] = None,
|
|
use_cache: bool = True,
|
|
rate_limit: bool = True,
|
|
rate_delay: float = 1.0
|
|
) -> List[CrawlResult]:
|
|
"""
|
|
批量 GET 请求（并发）
|
|
|
|
Args:
|
|
urls: URL 列表
|
|
headers: 额外 headers
|
|
use_cache: 是否使用缓存
|
|
rate_limit: 是否限流
|
|
rate_delay: 限流延迟（秒）
|
|
|
|
Returns:
|
|
List[CrawlResult]
|
|
"""
|
|
if not urls:
|
|
return []
|
|
|
|
results = [None] * len(urls)
|
|
|
|
def fetch_with_index(index: int, url: str) -> Tuple[int, CrawlResult]:
|
|
if rate_limit:
|
|
self._rate_limit(url, rate_delay)
|
|
result = self._single_get(url, headers, use_cache)
|
|
return index, result
|
|
|
|
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
|
|
futures = {
|
|
executor.submit(fetch_with_index, i, url): i
|
|
for i, url in enumerate(urls)
|
|
}
|
|
|
|
for future in as_completed(futures):
|
|
try:
|
|
index, result = future.result()
|
|
results[index] = result
|
|
except Exception as e:
|
|
index = futures[future]
|
|
results[index] = CrawlResult(
|
|
url=urls[index],
|
|
success=False,
|
|
error=str(e),
|
|
)
|
|
|
|
return results
|
|
|
|
def get_stats(self) -> Dict[str, Any]:
|
|
"""获取统计信息"""
|
|
with self._stats_lock:
|
|
stats = dict(self._stats)
|
|
|
|
stats['cache_size'] = len(self._cache)
|
|
stats['cache_hit_rate'] = (
|
|
stats['cache_hits'] / stats['total_requests']
|
|
if stats['total_requests'] > 0 else 0
|
|
)
|
|
stats['avg_response_time'] = (
|
|
stats['total_time'] / stats['total_requests']
|
|
if stats['total_requests'] > 0 else 0
|
|
)
|
|
|
|
return stats
|
|
|
|
def clear_cache(self):
|
|
"""清除缓存"""
|
|
with self._cache_lock:
|
|
self._cache.clear()
|
|
|
|
def reset_stats(self):
|
|
"""重置统计"""
|
|
with self._stats_lock:
|
|
self._stats = {
|
|
'total_requests': 0,
|
|
'cache_hits': 0,
|
|
'errors': 0,
|
|
'total_time': 0.0,
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# 便捷函数
|
|
# ============================================================
|
|
|
|
_global_crawler: Optional[PerformanceCrawler] = None
|
|
|
|
|
|
def get_crawler(max_workers: int = 5, **kwargs) -> PerformanceCrawler:
|
|
"""获取全局爬虫实例"""
|
|
global _global_crawler
|
|
if _global_crawler is None:
|
|
_global_crawler = PerformanceCrawler(max_workers=max_workers, **kwargs)
|
|
return _global_crawler
|
|
|
|
|
|
def fast_get(url: str, **kwargs) -> CrawlResult:
|
|
"""快速 GET 请求"""
|
|
return get_crawler().get(url, **kwargs)
|
|
|
|
|
|
def fast_batch_get(urls: List[str], **kwargs) -> List[CrawlResult]:
|
|
"""快速批量 GET 请求"""
|
|
return get_crawler().batch_get(urls, **kwargs)
|