Python爬虫第13课：爬虫性能优化与监控

课程目标

• 掌握爬虫性能瓶颈分析方法
• 学习系统监控与告警机制
• 实现资源使用优化策略
• 构建高可用爬虫架构

1. 性能瓶颈分析

1.1 性能指标监控

# performance_monitor.py
import time
import psutil
import threading
import logging
from typing importDict, List, Any, Optional
from dataclasses import dataclass, asdict
from collections import deque
import json
import statistics
from datetime import datetime, timedelta

@dataclass
classPerformanceMetrics:
"""性能指标"""
    timestamp: float
    cpu_percent: float
    memory_percent: float
    memory_used_mb: float
    disk_io_read_mb: float
    disk_io_write_mb: float
    network_sent_mb: float
    network_recv_mb: float
    active_threads: int
    requests_per_second: float
    response_time_avg: float
    error_rate: float
    queue_size: int = 0

defto_dict(self) -> Dict[str, Any]:
return asdict(self)

classSystemMonitor:
"""系统性能监控器"""

def__init__(self, collection_interval: float = 1.0, max_history: int = 3600):
self.collection_interval = collection_interval
self.max_history = max_history
self.metrics_history = deque(maxlen=max_history)
self.running = False
self.monitor_thread = None
self.logger = logging.getLogger(__name__)

# 网络和磁盘IO基准值
self.last_net_io = psutil.net_io_counters()
self.last_disk_io = psutil.disk_io_counters()
self.last_check_time = time.time()

# 请求统计
self.request_count = 0
self.response_times = deque(maxlen=1000)
self.error_count = 0
self.last_request_reset = time.time()

defrecord_request(self, response_time: float, is_error: bool = False):
"""记录请求"""
self.request_count += 1
self.response_times.append(response_time)
if is_error:
self.error_count += 1

defcollect_metrics(self) -> PerformanceMetrics:
"""收集性能指标"""
        current_time = time.time()

# CPU和内存
        cpu_percent = psutil.cpu_percent()
        memory = psutil.virtual_memory()
        memory_percent = memory.percent
        memory_used_mb = memory.used / (1024 * 1024)

# 磁盘IO
        current_disk_io = psutil.disk_io_counters()
        time_delta = current_time - self.last_check_time

if time_delta > 0:
            disk_read_mb = (current_disk_io.read_bytes - self.last_disk_io.read_bytes) / (1024 * 1024) / time_delta
            disk_write_mb = (current_disk_io.write_bytes - self.last_disk_io.write_bytes) / (1024 * 1024) / time_delta
else:
            disk_read_mb = disk_write_mb = 0

# 网络IO
        current_net_io = psutil.net_io_counters()
if time_delta > 0:
            net_sent_mb = (current_net_io.bytes_sent - self.last_net_io.bytes_sent) / (1024 * 1024) / time_delta
            net_recv_mb = (current_net_io.bytes_recv - self.last_net_io.bytes_recv) / (1024 * 1024) / time_delta
else:
            net_sent_mb = net_recv_mb = 0

# 线程数
        active_threads = threading.active_count()

# 请求统计
        request_time_delta = current_time - self.last_request_reset
if request_time_delta > 0:
            requests_per_second = self.request_count / request_time_delta
else:
            requests_per_second = 0

# 响应时间
        response_time_avg = statistics.mean(self.response_times) ifself.response_times else0

# 错误率
        error_rate = (self.error_count / self.request_count) ifself.request_count > 0else0

# 更新基准值
self.last_disk_io = current_disk_io
self.last_net_io = current_net_io
self.last_check_time = current_time

# 重置请求统计（每分钟重置一次）
if request_time_delta >= 60:
self.request_count = 0
self.error_count = 0
self.last_request_reset = current_time

return PerformanceMetrics(
            timestamp=current_time,
            cpu_percent=cpu_percent,
            memory_percent=memory_percent,
            memory_used_mb=memory_used_mb,
            disk_io_read_mb=disk_read_mb,
            disk_io_write_mb=disk_write_mb,
            network_sent_mb=net_sent_mb,
            network_recv_mb=net_recv_mb,
            active_threads=active_threads,
            requests_per_second=requests_per_second,
            response_time_avg=response_time_avg,
            error_rate=error_rate
        )

defmonitor_loop(self):
"""监控循环"""
whileself.running:
try:
                metrics = self.collect_metrics()
self.metrics_history.append(metrics)

# 检查异常情况
self.check_alerts(metrics)

                time.sleep(self.collection_interval)

except Exception as e:
self.logger.error(f"监控循环错误: {e}")
                time.sleep(self.collection_interval)

defcheck_alerts(self, metrics: PerformanceMetrics):
"""检查告警条件"""
        alerts = []

# CPU使用率告警
if metrics.cpu_percent > 80:
            alerts.append(f"CPU使用率过高: {metrics.cpu_percent:.1f}%")

# 内存使用率告警
if metrics.memory_percent > 85:
            alerts.append(f"内存使用率过高: {metrics.memory_percent:.1f}%")

# 响应时间告警
if metrics.response_time_avg > 5.0:
            alerts.append(f"平均响应时间过长: {metrics.response_time_avg:.2f}s")

# 错误率告警
if metrics.error_rate > 0.1:
            alerts.append(f"错误率过高: {metrics.error_rate:.2%}")

# 线程数告警
if metrics.active_threads > 100:
            alerts.append(f"活跃线程数过多: {metrics.active_threads}")

for alert in alerts:
self.logger.warning(f"性能告警: {alert}")

defstart(self):
"""启动监控"""
ifself.running:
return

self.running = True
self.monitor_thread = threading.Thread(target=self.monitor_loop)
self.monitor_thread.start()
self.logger.info("性能监控已启动")

defstop(self):
"""停止监控"""
self.running = False
ifself.monitor_thread:
self.monitor_thread.join()
self.logger.info("性能监控已停止")

defget_current_metrics(self) -> Optional[PerformanceMetrics]:
"""获取当前指标"""
ifself.metrics_history:
returnself.metrics_history[-1]
returnNone

defget_metrics_summary(self, minutes: int = 10) -> Dict[str, Any]:
"""获取指标摘要"""
ifnotself.metrics_history:
return {}

# 获取最近N分钟的数据
        cutoff_time = time.time() - (minutes * 60)
        recent_metrics = [m for m inself.metrics_history if m.timestamp >= cutoff_time]

ifnot recent_metrics:
return {}

# 计算统计信息
        cpu_values = [m.cpu_percent for m in recent_metrics]
        memory_values = [m.memory_percent for m in recent_metrics]
        response_times = [m.response_time_avg for m in recent_metrics if m.response_time_avg > 0]
        error_rates = [m.error_rate for m in recent_metrics]

return {
'time_range_minutes': minutes,
'sample_count': len(recent_metrics),
'cpu': {
'avg': statistics.mean(cpu_values),
'max': max(cpu_values),
'min': min(cpu_values)
            },
'memory': {
'avg': statistics.mean(memory_values),
'max': max(memory_values),
'min': min(memory_values)
            },
'response_time': {
'avg': statistics.mean(response_times) if response_times else0,
'max': max(response_times) if response_times else0,
'min': min(response_times) if response_times else0
            },
'error_rate': {
'avg': statistics.mean(error_rates),
'max': max(error_rates),
'min': min(error_rates)
            }
        }

defexport_metrics(self, filename: str, hours: int = 1):
"""导出指标数据"""
        cutoff_time = time.time() - (hours * 3600)
        export_data = [
            m.to_dict() for m inself.metrics_history 
if m.timestamp >= cutoff_time
        ]

withopen(filename, 'w', encoding='utf-8') as f:
            json.dump(export_data, f, indent=2)

self.logger.info(f"导出了 {len(export_data)} 条指标数据到 {filename}")

classBottleneckAnalyzer:
"""性能瓶颈分析器"""

def__init__(self, monitor: SystemMonitor):
self.monitor = monitor
self.logger = logging.getLogger(__name__)

defanalyze_bottlenecks(self, analysis_minutes: int = 30) -> Dict[str, Any]:
"""分析性能瓶颈"""
        summary = self.monitor.get_metrics_summary(analysis_minutes)

ifnot summary:
return {'error': '没有足够的监控数据'}

        bottlenecks = []
        recommendations = []

# CPU瓶颈分析
if summary['cpu']['avg'] > 70:
            bottlenecks.append({
'type': 'CPU',
'severity': 'high'if summary['cpu']['avg'] > 85else'medium',
'description': f"CPU平均使用率 {summary['cpu']['avg']:.1f}%",
'max_value': summary['cpu']['max']
            })
            recommendations.extend([
"考虑增加并发控制，减少同时运行的任务数",
"优化CPU密集型操作，如数据解析和处理",
"使用多进程而非多线程处理CPU密集型任务"
            ])

# 内存瓶颈分析
if summary['memory']['avg'] > 75:
            bottlenecks.append({
'type': 'Memory',
'severity': 'high'if summary['memory']['avg'] > 90else'medium',
'description': f"内存平均使用率 {summary['memory']['avg']:.1f}%",
'max_value': summary['memory']['max']
            })
            recommendations.extend([
"实现数据流式处理，避免大量数据同时加载到内存",
"优化数据结构，减少内存占用",
"增加内存回收机制，及时释放不用的对象"
            ])

# 响应时间瓶颈分析
if summary['response_time']['avg'] > 3.0:
            bottlenecks.append({
'type': 'Response Time',
'severity': 'high'if summary['response_time']['avg'] > 5.0else'medium',
'description': f"平均响应时间 {summary['response_time']['avg']:.2f}s",
'max_value': summary['response_time']['max']
            })
            recommendations.extend([
"优化网络请求，使用连接池和Keep-Alive",
"实现请求缓存，避免重复请求",
"调整超时设置，避免长时间等待"
            ])

# 错误率瓶颈分析
if summary['error_rate']['avg'] > 0.05:
            bottlenecks.append({
'type': 'Error Rate',
'severity': 'high'if summary['error_rate']['avg'] > 0.1else'medium',
'description': f"平均错误率 {summary['error_rate']['avg']:.2%}",
'max_value': summary['error_rate']['max']
            })
            recommendations.extend([
"增强错误处理和重试机制",
"分析错误原因，优化请求策略",
"实现智能退避算法，避免频繁失败"
            ])

# 综合评分
        total_score = 100
for bottleneck in bottlenecks:
if bottleneck['severity'] == 'high':
                total_score -= 25
elif bottleneck['severity'] == 'medium':
                total_score -= 15

        performance_grade = 'A'if total_score >= 90else'B'if total_score >= 75else'C'if total_score >= 60else'D'

return {
'analysis_time': datetime.now().isoformat(),
'analysis_period_minutes': analysis_minutes,
'performance_score': total_score,
'performance_grade': performance_grade,
'bottlenecks': bottlenecks,
'recommendations': recommendations,
'summary': summary
        }

defgenerate_optimization_plan(self, analysis_result: Dict[str, Any]) -> Dict[str, Any]:
"""生成优化计划"""
        bottlenecks = analysis_result.get('bottlenecks', [])

# 按严重程度排序
        high_priority = [b for b in bottlenecks if b['severity'] == 'high']
        medium_priority = [b for b in bottlenecks if b['severity'] == 'medium']

        optimization_tasks = []

# 高优先级任务
for bottleneck in high_priority:
if bottleneck['type'] == 'CPU':
                optimization_tasks.append({
'priority': 'high',
'task': 'CPU优化',
'actions': [
'实现智能并发控制',
'优化数据处理算法',
'使用异步IO替代同步操作'
                    ],
'expected_improvement': '20-30% CPU使用率降低'
                })

elif bottleneck['type'] == 'Memory':
                optimization_tasks.append({
'priority': 'high',
'task': '内存优化',
'actions': [
'实现流式数据处理',
'优化对象生命周期管理',
'增加内存监控和回收'
                    ],
'expected_improvement': '15-25% 内存使用率降低'
                })

# 中等优先级任务
for bottleneck in medium_priority:
if bottleneck['type'] == 'Response Time':
                optimization_tasks.append({
'priority': 'medium',
'task': '响应时间优化',
'actions': [
'实现请求缓存机制',
'优化网络连接配置',
'增加CDN和代理支持'
                    ],
'expected_improvement': '30-50% 响应时间改善'
                })

return {
'plan_created': datetime.now().isoformat(),
'total_tasks': len(optimization_tasks),
'high_priority_tasks': len([t for t in optimization_tasks if t['priority'] == 'high']),
'medium_priority_tasks': len([t for t in optimization_tasks if t['priority'] == 'medium']),
'optimization_tasks': optimization_tasks,
'estimated_completion_time': '1-2 weeks'
        }

# 使用示例
if __name__ == "__main__":
import random

# 配置日志
    logging.basicConfig(level=logging.INFO)

# 创建系统监控器
    monitor = SystemMonitor(collection_interval=1.0)
    monitor.start()

# 模拟一些请求
for i inrange(100):
        response_time = random.uniform(0.5, 3.0)
        is_error = random.random() < 0.05
        monitor.record_request(response_time, is_error)
        time.sleep(0.1)

# 等待收集一些数据
    time.sleep(10)

# 创建瓶颈分析器
    analyzer = BottleneckAnalyzer(monitor)

# 分析瓶颈
    analysis = analyzer.analyze_bottlenecks(analysis_minutes=1)
print(f"瓶颈分析结果: {json.dumps(analysis, indent=2, ensure_ascii=False)}")

# 生成优化计划
    plan = analyzer.generate_optimization_plan(analysis)
print(f"优化计划: {json.dumps(plan, indent=2, ensure_ascii=False)}")

# 导出指标
    monitor.export_metrics("performance_metrics.json", hours=1)

# 停止监控
    monitor.stop()

1.2 代码性能分析

# code_profiler.py
import cProfile
import pstats
import io
import time
import functools
import threading
from typing importDict, Any, Callable, Optional
import logging
from contextlib import contextmanager

classCodeProfiler:
"""代码性能分析器"""

def__init__(self):
self.profiles = {}
self.logger = logging.getLogger(__name__)

defprofile_function(self, func_name: str = None):
"""函数性能分析装饰器"""
defdecorator(func: Callable) -> Callable:
            name = func_name orf"{func.__module__}.{func.__name__}"

            @functools.wraps(func)
defwrapper(*args, **kwargs):
                profiler = cProfile.Profile()
                profiler.enable()

try:
                    result = func(*args, **kwargs)
return result
finally:
                    profiler.disable()

# 保存分析结果
                    s = io.StringIO()
                    ps = pstats.Stats(profiler, stream=s)
                    ps.sort_stats('cumulative')
                    ps.print_stats()

self.profiles[name] = {
'timestamp': time.time(),
'stats': s.getvalue(),
'function': name
                    }

return wrapper
return decorator

    @contextmanager
defprofile_block(self, block_name: str):
"""代码块性能分析上下文管理器"""
        profiler = cProfile.Profile()
        profiler.enable()

try:
yield
finally:
            profiler.disable()

            s = io.StringIO()
            ps = pstats.Stats(profiler, stream=s)
            ps.sort_stats('cumulative')
            ps.print_stats()

self.profiles[block_name] = {
'timestamp': time.time(),
'stats': s.getvalue(),
'block': block_name
            }

defget_profile_report(self, name: str) -> Optional[str]:
"""获取性能分析报告"""
if name inself.profiles:
returnself.profiles[name]['stats']
returnNone

defget_all_profiles(self) -> Dict[str, Any]:
"""获取所有性能分析结果"""
returnself.profiles

defclear_profiles(self):
"""清除所有分析结果"""
self.profiles.clear()

classTimingProfiler:
"""时间性能分析器"""

def__init__(self):
self.timings = {}
self.lock = threading.Lock()
self.logger = logging.getLogger(__name__)

deftime_function(self, func_name: str = None):
"""函数计时装饰器"""
defdecorator(func: Callable) -> Callable:
            name = func_name orf"{func.__module__}.{func.__name__}"

            @functools.wraps(func)
defwrapper(*args, **kwargs):
                start_time = time.time()

try:
                    result = func(*args, **kwargs)
return result
finally:
                    end_time = time.time()
                    duration = end_time - start_time

withself.lock:
if name notinself.timings:
self.timings[name] = []
self.timings[name].append({
'duration': duration,
'timestamp': start_time,
'args_count': len(args),
'kwargs_count': len(kwargs)
                        })

return wrapper
return decorator

    @contextmanager
deftime_block(self, block_name: str):
"""代码块计时上下文管理器"""
        start_time = time.time()

try:
yield
finally:
            end_time = time.time()
            duration = end_time - start_time

withself.lock:
if block_name notinself.timings:
self.timings[block_name] = []
self.timings[block_name].append({
'duration': duration,
'timestamp': start_time
                })

defget_timing_stats(self, name: str) -> Optional[Dict[str, Any]]:
"""获取计时统计"""
withself.lock:
if name notinself.timings:
returnNone

            durations = [t['duration'] for t inself.timings[name]]

ifnot durations:
returnNone

return {
'count': len(durations),
'total_time': sum(durations),
'avg_time': sum(durations) / len(durations),
'min_time': min(durations),
'max_time': max(durations),
'last_call': max(t['timestamp'] for t inself.timings[name])
            }

defget_all_stats(self) -> Dict[str, Dict[str, Any]]:
"""获取所有计时统计"""
        stats = {}
withself.lock:
for name inself.timings:
                stats[name] = self.get_timing_stats(name)
return stats

defgenerate_report(self) -> str:
"""生成性能报告"""
        stats = self.get_all_stats()

ifnot stats:
return"没有性能数据"

        report_lines = ["性能分析报告", "=" * 50]

# 按平均时间排序
        sorted_stats = sorted(
            stats.items(), 
            key=lambda x: x[1]['avg_time'] if x[1] else0, 
            reverse=True
        )

for name, stat in sorted_stats:
if stat:
                report_lines.extend([
f"\n函数/代码块: {name}",
f"  调用次数: {stat['count']}",
f"  总时间: {stat['total_time']:.4f}s",
f"  平均时间: {stat['avg_time']:.4f}s",
f"  最短时间: {stat['min_time']:.4f}s",
f"  最长时间: {stat['max_time']:.4f}s"
                ])

return"\n".join(report_lines)

# 使用示例
profiler = CodeProfiler()
timer = TimingProfiler()

@profiler.profile_function("example_function")
@timer.time_function("example_function")
defexample_function(n: int) -> int:
"""示例函数"""
    total = 0
for i inrange(n):
        total += i * i
return total

@timer.time_function("slow_function")
defslow_function():
"""慢函数示例"""
    time.sleep(0.1)
return"done"

if __name__ == "__main__":
# 测试函数
for i inrange(10):
        example_function(1000)
        slow_function()

# 使用代码块分析
with profiler.profile_block("data_processing"):
with timer.time_block("data_processing"):
            data = [i ** 2for i inrange(10000)]
            result = sum(data)

# 生成报告
print(timer.generate_report())

# 获取详细分析
print("\n详细性能分析:")
for name in profiler.get_all_profiles():
print(f"\n{name}:")
print(profiler.get_profile_report(name)[:500] + "...")

2. 资源使用优化

2.1 内存优化策略

# memory_optimizer.py
import gc
import sys
import weakref
import threading
import time
import logging
from typing importDict, Any, List, Optional, Generator
from collections import defaultdict
import tracemalloc
from dataclasses import dataclass

@dataclass
classMemorySnapshot:
"""内存快照"""
    timestamp: float
    current_mb: float
    peak_mb: float
    traced_mb: float
    object_count: int
    gc_collections: Dict[int, int]

classMemoryOptimizer:
"""内存优化器"""

def__init__(self, enable_tracing: bool = True):
self.enable_tracing = enable_tracing
self.snapshots = []
self.object_tracker = defaultdict(int)
self.weak_refs = []
self.logger = logging.getLogger(__name__)

if enable_tracing:
            tracemalloc.start()

# 启动内存监控
self.monitoring = False
self.monitor_thread = None

deftake_snapshot(self) -> MemorySnapshot:
"""拍摄内存快照"""
# 基本内存信息
import psutil
        process = psutil.Process()
        memory_info = process.memory_info()
        current_mb = memory_info.rss / (1024 * 1024)

# 获取峰值内存
        peak_mb = process.memory_info().peak_wss / (1024 * 1024) ifhasattr(process.memory_info(), 'peak_wss') else current_mb

# tracemalloc信息
        traced_mb = 0
ifself.enable_tracing:
            current, peak = tracemalloc.get_traced_memory()
            traced_mb = current / (1024 * 1024)

# 对象计数
        object_count = len(gc.get_objects())

# GC统计
        gc_stats = {}
for i inrange(3):
            gc_stats[i] = gc.get_count()[i]

        snapshot = MemorySnapshot(
            timestamp=time.time(),
            current_mb=current_mb,
            peak_mb=peak_mb,
            traced_mb=traced_mb,
            object_count=object_count,
            gc_collections=gc_stats
        )

self.snapshots.append(snapshot)
return snapshot

defanalyze_memory_growth(self, window_size: int = 10) -> Dict[str, Any]:
"""分析内存增长趋势"""
iflen(self.snapshots) < 2:
return {'error': '快照数据不足'}

        recent_snapshots = self.snapshots[-window_size:]

# 计算增长率
        first_snapshot = recent_snapshots[0]
        last_snapshot = recent_snapshots[-1]

        time_delta = last_snapshot.timestamp - first_snapshot.timestamp
        memory_delta = last_snapshot.current_mb - first_snapshot.current_mb

        growth_rate = memory_delta / time_delta if time_delta > 0else0

# 检测内存泄漏
        is_leaking = growth_rate > 1.0# 每秒增长超过1MB

return {
'window_size': len(recent_snapshots),
'time_span_seconds': time_delta,
'memory_growth_mb': memory_delta,
'growth_rate_mb_per_second': growth_rate,
'is_potential_leak': is_leaking,
'current_memory_mb': last_snapshot.current_mb,
'object_count_change': last_snapshot.object_count - first_snapshot.object_count
        }

defget_top_memory_objects(self, limit: int = 10) -> List[Dict[str, Any]]:
"""获取占用内存最多的对象类型"""
ifnotself.enable_tracing:
return []

        snapshot = tracemalloc.take_snapshot()
        top_stats = snapshot.statistics('lineno')

        result = []
for index, stat inenumerate(top_stats[:limit]):
            result.append({
'rank': index + 1,
'filename': stat.traceback.format()[0] if stat.traceback.format() else'unknown',
'size_mb': stat.size / (1024 * 1024),
'count': stat.count
            })

return result

defforce_garbage_collection(self) -> Dict[str, int]:
"""强制垃圾回收"""
        before_objects = len(gc.get_objects())

# 执行垃圾回收
        collected = {}
for generation inrange(3):
            collected[generation] = gc.collect(generation)

        after_objects = len(gc.get_objects())
        freed_objects = before_objects - after_objects

self.logger.info(f"垃圾回收完成，释放了 {freed_objects} 个对象")

return {
'freed_objects': freed_objects,
'collections_by_generation': collected,
'remaining_objects': after_objects
        }

defoptimize_memory_usage(self) -> Dict[str, Any]:
"""优化内存使用"""
        before_snapshot = self.take_snapshot()

# 1. 强制垃圾回收
        gc_result = self.force_garbage_collection()

# 2. 清理弱引用
        cleaned_refs = 0
for ref inself.weak_refs[:]:
if ref() isNone:
self.weak_refs.remove(ref)
                cleaned_refs += 1

# 3. 清理缓存（如果有的话）
# 这里可以添加特定的缓存清理逻辑

        after_snapshot = self.take_snapshot()

        memory_saved = before_snapshot.current_mb - after_snapshot.current_mb

return {
'memory_before_mb': before_snapshot.current_mb,
'memory_after_mb': after_snapshot.current_mb,
'memory_saved_mb': memory_saved,
'gc_result': gc_result,
'cleaned_weak_refs': cleaned_refs,
'optimization_effective': memory_saved > 0
        }

defstart_monitoring(self, interval: float = 30.0):
"""启动内存监控"""
ifself.monitoring:
return

self.monitoring = True

defmonitor_loop():
whileself.monitoring:
try:
                    snapshot = self.take_snapshot()

# 检查内存增长
iflen(self.snapshots) >= 5:
                        analysis = self.analyze_memory_growth(5)
if analysis.get('is_potential_leak', False):
self.logger.warning(f"检测到潜在内存泄漏: {analysis['growth_rate_mb_per_second']:.2f} MB/s")

# 自动优化
                            opt_result = self.optimize_memory_usage()
self.logger.info(f"自动内存优化，节省 {opt_result['memory_saved_mb']:.2f} MB")

                    time.sleep(interval)

except Exception as e:
self.logger.error(f"内存监控错误: {e}")
                    time.sleep(interval)

self.monitor_thread = threading.Thread(target=monitor_loop)
self.monitor_thread.start()
self.logger.info("内存监控已启动")

defstop_monitoring(self):
"""停止内存监控"""
self.monitoring = False
ifself.monitor_thread:
self.monitor_thread.join()
self.logger.info("内存监控已停止")

defregister_object(self, obj: Any, name: str = None):
"""注册对象用于跟踪"""
        obj_type = type(obj).__name__
if name:
            obj_type = f"{obj_type}({name})"

self.object_tracker[obj_type] += 1

# 创建弱引用
defcleanup_callback(ref):
self.object_tracker[obj_type] -= 1

        weak_ref = weakref.ref(obj, cleanup_callback)
self.weak_refs.append(weak_ref)

return weak_ref

classStreamProcessor:
"""流式数据处理器 - 内存优化示例"""

def__init__(self, chunk_size: int = 1000):
self.chunk_size = chunk_size
self.memory_optimizer = MemoryOptimizer()
self.logger = logging.getLogger(__name__)

defprocess_large_dataset(self, data_generator: Generator) -> Generator:
"""流式处理大数据集"""
        chunk = []

for item in data_generator:
            chunk.append(item)

iflen(chunk) >= self.chunk_size:
# 处理当前块
                processed_chunk = self._process_chunk(chunk)

# 返回处理结果
for result in processed_chunk:
yield result

# 清理内存
                chunk.clear()

# 定期检查内存使用
iflen(chunk) % (self.chunk_size * 10) == 0:
                    snapshot = self.memory_optimizer.take_snapshot()
if snapshot.current_mb > 500:  # 超过500MB时优化
self.memory_optimizer.optimize_memory_usage()

# 处理剩余数据
if chunk:
            processed_chunk = self._process_chunk(chunk)
for result in processed_chunk:
yield result

def_process_chunk(self, chunk: List[Any]) -> List[Any]:
"""处理数据块"""
# 这里实现具体的数据处理逻辑
return [self._process_item(item) for item in chunk]

def_process_item(self, item: Any) -> Any:
"""处理单个数据项"""
# 示例处理逻辑
returnstr(item).upper()

# 使用示例
if __name__ == "__main__":
import random

# 配置日志
    logging.basicConfig(level=logging.INFO)

# 创建内存优化器
    optimizer = MemoryOptimizer()
    optimizer.start_monitoring(interval=5.0)

try:
# 模拟内存使用
        large_data = []
for i inrange(10000):
            large_data.append([random.random() for _ inrange(100)])

if i % 1000 == 0:
                snapshot = optimizer.take_snapshot()
print(f"当前内存使用: {snapshot.current_mb:.2f} MB")

# 分析内存增长
        analysis = optimizer.analyze_memory_growth()
print(f"内存增长分析: {analysis}")

# 获取内存占用最多的对象
        top_objects = optimizer.get_top_memory_objects()
print(f"内存占用最多的对象: {top_objects}")

# 优化内存
        opt_result = optimizer.optimize_memory_usage()
print(f"内存优化结果: {opt_result}")

# 测试流式处理
defdata_generator():
for i inrange(100000):
yieldf"data_item_{i}"

        processor = StreamProcessor(chunk_size=1000)
        processed_count = 0

for processed_item in processor.process_large_dataset(data_generator()):
            processed_count += 1
if processed_count % 10000 == 0:
print(f"已处理 {processed_count} 个项目")

print(f"流式处理完成，总共处理 {processed_count} 个项目")

finally:
        optimizer.stop_monitoring()

2.2 连接池优化

# connection_pool.py
import asyncio
import aiohttp
import time
import threading
import logging
from typing importDict, Any, Optional, List
from dataclasses import dataclass, field
from collections import deque
import weakref
import ssl

@dataclass
classConnectionStats:
"""连接统计信息"""
    total_created: int = 0
    total_closed: int = 0
    active_connections: int = 0
    pool_hits: int = 0
    pool_misses: int = 0
    average_lifetime: float = 0.0
    peak_connections: int = 0

defhit_rate(self) -> float:
        total_requests = self.pool_hits + self.pool_misses
returnself.pool_hits / total_requests if total_requests > 0else0.0

@dataclass
classPooledConnection:
"""池化连接"""
    session: aiohttp.ClientSession
    created_time: float = field(default_factory=time.time)
    last_used: float = field(default_factory=time.time)
    use_count: int = 0
    is_active: bool = True

defmark_used(self):
"""标记连接被使用"""
self.last_used = time.time()
self.use_count += 1

defis_expired(self, max_lifetime: float, max_idle: float) -> bool:
"""检查连接是否过期"""
        now = time.time()
        lifetime_expired = (now - self.created_time) > max_lifetime
        idle_expired = (now - self.last_used) > max_idle
return lifetime_expired or idle_expired

classOptimizedConnectionPool:
"""优化的连接池"""

def__init__(
        self,
        max_connections: int = 100,
        max_connections_per_host: int = 30,
        max_lifetime: float = 3600.0,  # 1小时
        max_idle_time: float = 300.0,  # 5分钟
        cleanup_interval: float = 60.0,  # 1分钟清理一次
        enable_ssl: bool = True
):
self.max_connections = max_connections
self.max_connections_per_host = max_connections_per_host
self.max_lifetime = max_lifetime
self.max_idle_time = max_idle_time
self.cleanup_interval = cleanup_interval
self.enable_ssl = enable_ssl

# 连接池
self.pools: Dict[str, deque] = {}
self.stats = ConnectionStats()
self.lock = asyncio.Lock()

# 清理任务
self.cleanup_task = None
self.running = False

# SSL配置
self.ssl_context = self._create_ssl_context() if enable_ssl elseNone

self.logger = logging.getLogger(__name__)

def_create_ssl_context(self) -> ssl.SSLContext:
"""创建SSL上下文"""
        context = ssl.create_default_context()
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
return context

def_get_pool_key(self, url: str) -> str:
"""获取连接池键"""
from urllib.parse import urlparse
        parsed = urlparse(url)
returnf"{parsed.scheme}://{parsed.netloc}"

asyncdefget_session(self, url: str) -> aiohttp.ClientSession:
"""获取会话连接"""
        pool_key = self._get_pool_key(url)

asyncwithself.lock:
# 检查是否有可用连接
if pool_key inself.pools andself.pools[pool_key]:
                connection = self.pools[pool_key].popleft()

# 检查连接是否过期
ifnot connection.is_expired(self.max_lifetime, self.max_idle_time):
                    connection.mark_used()
self.stats.pool_hits += 1
return connection.session
else:
# 连接过期，关闭它
await connection.session.close()
self.stats.total_closed += 1
self.stats.active_connections -= 1

# 创建新连接
self.stats.pool_misses += 1
returnawaitself._create_new_session(pool_key)

asyncdef_create_new_session(self, pool_key: str) -> aiohttp.ClientSession:
"""创建新的会话连接"""
# 检查连接数限制
ifself.stats.active_connections >= self.max_connections:
# 清理过期连接
awaitself._cleanup_expired_connections()

# 如果仍然超过限制，等待
ifself.stats.active_connections >= self.max_connections:
raise Exception("连接池已满，无法创建新连接")

# 创建连接器
        connector = aiohttp.TCPConnector(
            limit=self.max_connections,
            limit_per_host=self.max_connections_per_host,
            ttl_dns_cache=300,
            use_dns_cache=True,
            ssl=self.ssl_context,
            enable_cleanup_closed=True
        )

# 创建会话
        timeout = aiohttp.ClientTimeout(total=30, connect=10)
        session = aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={
'User-Agent': 'OptimizedCrawler/1.0',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive'
            }
        )

self.stats.total_created += 1
self.stats.active_connections += 1
self.stats.peak_connections = max(self.stats.peak_connections, self.stats.active_connections)

self.logger.debug(f"创建新连接: {pool_key}")
return session

asyncdefreturn_session(self, url: str, session: aiohttp.ClientSession):
"""归还会话连接"""
        pool_key = self._get_pool_key(url)

asyncwithself.lock:
if pool_key notinself.pools:
self.pools[pool_key] = deque()

# 检查池大小限制
iflen(self.pools[pool_key]) < self.max_connections_per_host:
                connection = PooledConnection(session=session)
self.pools[pool_key].append(connection)
else:
# 池已满，关闭连接
await session.close()
self.stats.total_closed += 1
self.stats.active_connections -= 1

asyncdef_cleanup_expired_connections(self):
"""清理过期连接"""
        cleaned_count = 0

for pool_key, pool inself.pools.items():
            expired_connections = []

# 找出过期连接
for connection in pool:
if connection.is_expired(self.max_lifetime, self.max_idle_time):
                    expired_connections.append(connection)

# 移除并关闭过期连接
for connection in expired_connections:
                pool.remove(connection)
await connection.session.close()
                cleaned_count += 1
self.stats.total_closed += 1
self.stats.active_connections -= 1

if cleaned_count > 0:
self.logger.info(f"清理了 {cleaned_count} 个过期连接")

asyncdef_cleanup_loop(self):
"""清理循环"""
whileself.running:
try:
awaitself._cleanup_expired_connections()
await asyncio.sleep(self.cleanup_interval)
except Exception as e:
self.logger.error(f"连接清理错误: {e}")
await asyncio.sleep(self.cleanup_interval)

asyncdefstart(self):
"""启动连接池"""
ifself.running:
return

self.running = True
self.cleanup_task = asyncio.create_task(self._cleanup_loop())
self.logger.info("连接池已启动")

asyncdefstop(self):
"""停止连接池"""
self.running = False

ifself.cleanup_task:
self.cleanup_task.cancel()
try:
awaitself.cleanup_task
except asyncio.CancelledError:
pass

# 关闭所有连接
asyncwithself.lock:
for pool inself.pools.values():
for connection in pool:
await connection.session.close()

self.pools.clear()
self.stats.active_connections = 0

self.logger.info("连接池已停止")

defget_stats(self) -> ConnectionStats:
"""获取连接池统计信息"""
returnself.stats

asyncdefhealth_check(self) -> Dict[str, Any]:
"""健康检查"""
asyncwithself.lock:
            total_pooled = sum(len(pool) for pool inself.pools.values())

return {
'running': self.running,
'total_pools': len(self.pools),
'total_pooled_connections': total_pooled,
'active_connections': self.stats.active_connections,
'hit_rate': self.stats.hit_rate(),
'peak_connections': self.stats.peak_connections,
'memory_usage_mb': self._estimate_memory_usage()
            }

def_estimate_memory_usage(self) -> float:
"""估算内存使用量"""
# 粗略估算每个连接占用约1MB内存
returnself.stats.active_connections * 1.0

classPooledHttpClient:
"""使用连接池的HTTP客户端"""

def__init__(self, connection_pool: OptimizedConnectionPool):
self.pool = connection_pool
self.logger = logging.getLogger(__name__)

asyncdefget(self, url: str, **kwargs) -> aiohttp.ClientResponse:
"""GET请求"""
        session = awaitself.pool.get_session(url)

try:
            response = await session.get(url, **kwargs)
return response
finally:
awaitself.pool.return_session(url, session)

asyncdefpost(self, url: str, **kwargs) -> aiohttp.ClientResponse:
"""POST请求"""
        session = awaitself.pool.get_session(url)

try:
            response = await session.post(url, **kwargs)
return response
finally:
awaitself.pool.return_session(url, session)

asyncdefrequest(self, method: str, url: str, **kwargs) -> aiohttp.ClientResponse:
"""通用请求方法"""
        session = awaitself.pool.get_session(url)

try:
            response = await session.request(method, url, **kwargs)
return response
finally:
awaitself.pool.return_session(url, session)

# 使用示例
asyncdefmain():
# 配置日志
    logging.basicConfig(level=logging.INFO)

# 创建连接池
    pool = OptimizedConnectionPool(
        max_connections=50,
        max_connections_per_host=10,
        max_lifetime=1800.0,  # 30分钟
        max_idle_time=300.0,  # 5分钟
        cleanup_interval=60.0# 1分钟清理一次
    )

# 启动连接池
await pool.start()

# 创建HTTP客户端
    client = PooledHttpClient(pool)

try:
# 测试请求
        test_urls = [
"https://httpbin.org/get",
"https://httpbin.org/json",
"https://httpbin.org/delay/1",
"https://httpbin.org/status/200"
        ]

# 并发请求测试
        tasks = []
for i inrange(20):
            url = test_urls[i % len(test_urls)]
            task = asyncio.create_task(client.get(url))
            tasks.append(task)

        responses = await asyncio.gather(*tasks, return_exceptions=True)

        success_count = sum(1for r in responses ifnotisinstance(r, Exception))
print(f"成功请求: {success_count}/{len(responses)}")

# 获取连接池统计
        stats = pool.get_stats()
print(f"连接池统计: 命中率={stats.hit_rate():.2%}, 活跃连接={stats.active_connections}")

# 健康检查
        health = await pool.health_check()
print(f"健康检查: {health}")

finally:
# 停止连接池
await pool.stop()

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

5. 实战案例：高性能新闻爬虫系统

5.1 系统架构设计

# news_crawler_system.py
import asyncio
import aiohttp
import time
import json
import logging
from typing importDict, Any, List, Optional
from dataclasses import dataclass, field
import hashlib
from urllib.parse import urljoin, urlparse
import re
from bs4 import BeautifulSoup

@dataclass
classNewsItem:
"""新闻条目"""
    url: str
    title: str = ""
    content: str = ""
    author: str = ""
    publish_time: str = ""
    category: str = ""
    tags: List[str] = field(default_factory=list)
    source: str = ""
    crawl_time: float = field(default_factory=time.time)

defto_dict(self) -> Dict[str, Any]:
return {
'url': self.url,
'title': self.title,
'content': self.content,
'author': self.author,
'publish_time': self.publish_time,
'category': self.category,
'tags': self.tags,
'source': self.source,
'crawl_time': self.crawl_time
        }

classNewsExtractor:
"""新闻内容提取器"""

def__init__(self):
self.logger = logging.getLogger(__name__)

# 常见新闻网站的选择器配置
self.site_configs = {
'default': {
'title': ['h1', '.title', '.headline', '[class*="title"]'],
'content': ['.content', '.article-content', '.post-content', '[class*="content"]'],
'author': ['.author', '.byline', '[class*="author"]'],
'time': ['.time', '.date', '.publish-time', '[class*="time"]', '[class*="date"]']
            },
'sina.com.cn': {
'title': ['.main-title', 'h1'],
'content': ['.article', '.content'],
'author': ['.author'],
'time': ['.time']
            },
'sohu.com': {
'title': ['.text-title', 'h1'],
'content': ['.text'],
'author': ['.author-name'],
'time': ['.time']
            }
        }

defget_site_config(self, url: str) -> Dict[str, List[str]]:
"""获取网站配置"""
        domain = urlparse(url).netloc
for site, config inself.site_configs.items():
if site in domain:
return config
returnself.site_configs['default']

defextract_text_by_selectors(self, soup: BeautifulSoup, selectors: List[str]) -> str:
"""通过选择器提取文本"""
for selector in selectors:
try:
                element = soup.select_one(selector)
if element:
return element.get_text(strip=True)
except Exception:
continue
return""

defextract_news(self, html: str, url: str) -> NewsItem:
"""提取新闻内容"""
try:
            soup = BeautifulSoup(html, 'html.parser')
            config = self.get_site_config(url)

# 提取标题
            title = self.extract_text_by_selectors(soup, config['title'])
ifnot title:
                title = soup.title.get_text(strip=True) if soup.title else""

# 提取内容
            content = self.extract_text_by_selectors(soup, config['content'])

# 提取作者
            author = self.extract_text_by_selectors(soup, config['author'])

# 提取时间
            publish_time = self.extract_text_by_selectors(soup, config['time'])

# 提取标签
            tags = []
            tag_elements = soup.select('.tag, .tags, [class*="tag"]')
for tag_elem in tag_elements:
                tag_text = tag_elem.get_text(strip=True)
if tag_text:
                    tags.append(tag_text)

# 确定来源
            source = urlparse(url).netloc

            news_item = NewsItem(
                url=url,
                title=title,
                content=content,
                author=author,
                publish_time=publish_time,
                tags=tags,
                source=source
            )

self.logger.debug(f"提取新闻成功: {title[:50]}...")
return news_item

except Exception as e:
self.logger.error(f"提取新闻失败 {url}: {e}")
return NewsItem(url=url)

classPerformanceOptimizedCrawler:
"""性能优化的爬虫"""

def__init__(self, 
                 max_concurrent: int = 100,
                 request_delay: float = 0.1,
                 timeout: float = 30.0):
self.max_concurrent = max_concurrent
self.request_delay = request_delay
self.timeout = timeout
self.session = None
self.semaphore = asyncio.Semaphore(max_concurrent)
self.extractor = NewsExtractor()
self.logger = logging.getLogger(__name__)

# 性能统计
self.stats = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'total_time': 0.0,
'avg_response_time': 0.0
        }

asyncdefcreate_session(self):
"""创建HTTP会话"""
        connector = aiohttp.TCPConnector(
            limit=self.max_concurrent,
            limit_per_host=20,
            ttl_dns_cache=300,
            use_dns_cache=True,
            keepalive_timeout=30,
            enable_cleanup_closed=True
        )

        timeout = aiohttp.ClientTimeout(total=self.timeout)

self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }
        )

asyncdefclose_session(self):
"""关闭HTTP会话"""
ifself.session:
awaitself.session.close()

asyncdeffetch_url(self, url: str) -> Optional[str]:
"""获取URL内容"""
asyncwithself.semaphore:
            start_time = time.time()

try:
await asyncio.sleep(self.request_delay)

asyncwithself.session.get(url) as response:
if response.status == 200:
                        content = await response.text()

# 更新统计
                        response_time = time.time() - start_time
self.stats['successful_requests'] += 1
self.stats['total_time'] += response_time

return content
else:
self.logger.warning(f"HTTP {response.status}: {url}")
self.stats['failed_requests'] += 1
returnNone

except Exception as e:
self.logger.error(f"请求失败 {url}: {e}")
self.stats['failed_requests'] += 1
returnNone

finally:
self.stats['total_requests'] += 1
ifself.stats['successful_requests'] > 0:
self.stats['avg_response_time'] = (
self.stats['total_time'] / self.stats['successful_requests']
                    )

asyncdefcrawl_news(self, url: str) -> Optional[NewsItem]:
"""爬取单个新闻"""
        html = awaitself.fetch_url(url)
if html:
returnself.extractor.extract_news(html, url)
returnNone

asyncdefcrawl_news_batch(self, urls: List[str]) -> List[NewsItem]:
"""批量爬取新闻"""
        tasks = [self.crawl_news(url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        news_items = []
for result in results:
ifisinstance(result, NewsItem):
                news_items.append(result)
elifisinstance(result, Exception):
self.logger.error(f"爬取异常: {result}")

return news_items

defget_performance_stats(self) -> Dict[str, Any]:
"""获取性能统计"""
        success_rate = 0.0
ifself.stats['total_requests'] > 0:
            success_rate = self.stats['successful_requests'] / self.stats['total_requests']

return {
'total_requests': self.stats['total_requests'],
'successful_requests': self.stats['successful_requests'],
'failed_requests': self.stats['failed_requests'],
'success_rate': success_rate,
'avg_response_time': self.stats['avg_response_time'],
'requests_per_second': (
self.stats['successful_requests'] / self.stats['total_time']
ifself.stats['total_time'] > 0else0
            )
        }

classNewsMonitoringSystem:
"""新闻监控系统"""

def__init__(self, crawler: PerformanceOptimizedCrawler):
self.crawler = crawler
self.monitoring_data = {
'start_time': time.time(),
'crawl_sessions': [],
'performance_history': [],
'error_log': []
        }
self.logger = logging.getLogger(__name__)

deflog_crawl_session(self, session_data: Dict[str, Any]):
"""记录爬取会话"""
        session_data['timestamp'] = time.time()
self.monitoring_data['crawl_sessions'].append(session_data)

# 保留最近100次会话
iflen(self.monitoring_data['crawl_sessions']) > 100:
self.monitoring_data['crawl_sessions'] = self.monitoring_data['crawl_sessions'][-100:]

deflog_performance(self):
"""记录性能数据"""
        perf_data = self.crawler.get_performance_stats()
        perf_data['timestamp'] = time.time()
self.monitoring_data['performance_history'].append(perf_data)

# 保留最近1000个数据点
iflen(self.monitoring_data['performance_history']) > 1000:
self.monitoring_data['performance_history'] = self.monitoring_data['performance_history'][-1000:]

deflog_error(self, error_msg: str, error_type: str = "general"):
"""记录错误"""
        error_data = {
'timestamp': time.time(),
'type': error_type,
'message': error_msg
        }
self.monitoring_data['error_log'].append(error_data)

# 保留最近500个错误
iflen(self.monitoring_data['error_log']) > 500:
self.monitoring_data['error_log'] = self.monitoring_data['error_log'][-500:]

defget_monitoring_report(self) -> Dict[str, Any]:
"""生成监控报告"""
        current_time = time.time()
        uptime = current_time - self.monitoring_data['start_time']

# 计算最近性能
        recent_perf = self.monitoring_data['performance_history'][-10:] ifself.monitoring_data['performance_history'] else []
        avg_success_rate = sum(p['success_rate'] for p in recent_perf) / len(recent_perf) if recent_perf else0
        avg_response_time = sum(p['avg_response_time'] for p in recent_perf) / len(recent_perf) if recent_perf else0

# 计算错误统计
        recent_errors = [e for e inself.monitoring_data['error_log'] if current_time - e['timestamp'] < 3600]  # 最近1小时

return {
'uptime_seconds': uptime,
'total_sessions': len(self.monitoring_data['crawl_sessions']),
'recent_avg_success_rate': avg_success_rate,
'recent_avg_response_time': avg_response_time,
'recent_errors_count': len(recent_errors),
'current_performance': self.crawler.get_performance_stats(),
'error_types': self._get_error_type_stats(recent_errors)
        }

def_get_error_type_stats(self, errors: List[Dict[str, Any]]) -> Dict[str, int]:
"""获取错误类型统计"""
        error_types = {}
for error in errors:
            error_type = error['type']
            error_types[error_type] = error_types.get(error_type, 0) + 1
return error_types

classHighPerformanceNewsSystem:
"""高性能新闻爬虫系统"""

def__init__(self, 
                 max_concurrent: int = 100,
                 request_delay: float = 0.1):
self.crawler = PerformanceOptimizedCrawler(
            max_concurrent=max_concurrent,
            request_delay=request_delay
        )
self.monitoring = NewsMonitoringSystem(self.crawler)
self.news_storage = []
self.logger = logging.getLogger(__name__)

asyncdefstart(self):
"""启动系统"""
awaitself.crawler.create_session()
self.logger.info("高性能新闻爬虫系统已启动")

asyncdefstop(self):
"""停止系统"""
awaitself.crawler.close_session()
self.logger.info("高性能新闻爬虫系统已停止")

asyncdefcrawl_news_sites(self, news_urls: List[str]) -> List[NewsItem]:
"""爬取新闻网站"""
        start_time = time.time()

try:
# 批量爬取
            news_items = awaitself.crawler.crawl_news_batch(news_urls)

# 过滤有效新闻
            valid_news = [item for item in news_items if item.title and item.content]

# 存储新闻
self.news_storage.extend(valid_news)

# 记录会话
            session_data = {
'urls_count': len(news_urls),
'valid_news_count': len(valid_news),
'duration': time.time() - start_time,
'performance': self.crawler.get_performance_stats()
            }
self.monitoring.log_crawl_session(session_data)
self.monitoring.log_performance()

self.logger.info(f"爬取完成: {len(valid_news)}/{len(news_urls)} 条有效新闻")
return valid_news

except Exception as e:
            error_msg = f"爬取新闻失败: {e}"
self.logger.error(error_msg)
self.monitoring.log_error(error_msg, "crawl_error")
return []

defget_system_status(self) -> Dict[str, Any]:
"""获取系统状态"""
return {
'news_count': len(self.news_storage),
'monitoring_report': self.monitoring.get_monitoring_report(),
'crawler_stats': self.crawler.get_performance_stats()
        }

defexport_news(self, format_type: str = "json") -> str:
"""导出新闻数据"""
if format_type == "json":
return json.dumps([item.to_dict() for item inself.news_storage], 
                            ensure_ascii=False, indent=2)
else:
raise ValueError(f"不支持的格式: {format_type}")

# 使用示例
asyncdefmain():
# 配置日志
    logging.basicConfig(
        level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

# 创建高性能新闻系统
    news_system = HighPerformanceNewsSystem(
        max_concurrent=50,
        request_delay=0.05
    )

await news_system.start()

try:
# 示例新闻URL列表
        news_urls = [
"https://news.sina.com.cn/c/2023-01-01/doc-example1.shtml",
"https://news.sohu.com/20230101/example2.shtml",
"https://www.163.com/news/article/example3.html",
# 添加更多新闻URL...
        ]

# 爬取新闻
        news_items = await news_system.crawl_news_sites(news_urls)

# 显示系统状态
        status = news_system.get_system_status()
print(f"系统状态: {json.dumps(status, indent=2, ensure_ascii=False)}")

# 导出新闻数据
if news_items:
            news_data = news_system.export_news()
print(f"导出新闻数据: {len(news_items)} 条")

# 保存到文件
withopen("news_data.json", "w", encoding="utf-8") as f:
                f.write(news_data)

finally:
await news_system.stop()

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