#Scrapy-Redis分布式架构 - 构建高性能分布式爬虫集群

"""
分布式爬虫核心优势：

1. 水平扩展能力：
   - 通过增加节点提升整体性能
   - 突破单机资源瓶颈
   - 支持弹性伸缩

2. 容错性：
   - 单节点故障不影响整体运行
   - 自动故障转移
   - 高可用保障

3. 负载均衡：
   - 任务自动分配
   - 资源利用率最大化
   - 避免单点过载

4. 数据一致性：
   - 统一去重机制
   - 避免重复抓取
   - 数据完整性保障
"""

"""
常见的分布式爬虫架构：

1. 中央队列模式：
   - Redis作为中央队列
   - 多个爬虫节点消费
   - 统一去重机制

2. 分片模式：
   - 任务预先分片
   - 每个节点负责特定分片
   - 减少协调开销

3. 混合模式：
   - 结合多种架构优势
   - 动态任务分配
   - 自适应调度
"""

# settings.py - Scrapy-Redis基础配置
import os

# 启用Redis调度器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 启用Redis去重过滤器
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# Redis连接配置
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379')

# 或者分别配置主机和端口
REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
REDIS_PASSWORD = os.getenv('REDIS_PASSWORD', None)
REDIS_DB = int(os.getenv('REDIS_DB', 0))

# 持久化配置 - 关闭时不清空队列
SCHEDULER_PERSIST = True

# 请求队列类型配置
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'  # 优先级队列（默认）
# SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.FifoQueue'    # FIFO队列
# SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.LifoQueue'    # LIFO队列

# 最大队列长度
SCHEDULER_QUEUE_KEY = '%(spider)s:requests'  # 队列键名格式
DUPEFILTER_KEY = '%(spider)s:dupefilter'     # 去重键名格式

# 清理Redis配置
REDIS_CLEAR_SCHEUDLER = True  # 启动时是否清空队列
REDIS_START_URLS_AS_SET = True  # 使用Set存储起始URL
REDIS_START_URLS_KEY = '%(name)s:start_urls'  # 起始URL键名

# 连接池配置
REDIS_PARAMS = {
    'socket_timeout': 30,
    'socket_connect_timeout': 30,
    'retry_on_timeout': True,
    'encoding': 'utf-8',
    'health_check_interval': 30,
    'max_connections': 20,
    'db': REDIS_DB,
    'password': REDIS_PASSWORD,
}

# 日志配置
LOG_LEVEL = 'INFO'

# advanced_settings.py - 高级配置选项
import redis

# 自定义Redis连接工厂
def get_redis_connection():
    """
    获取Redis连接
    """
    pool = redis.ConnectionPool(
        host=REDIS_HOST,
        port=REDIS_PORT,
        db=REDIS_DB,
        password=REDIS_PASSWORD,
        max_connections=20,
        health_check_interval=30
    )
    return redis.Redis(connection_pool=pool)

# 性能优化配置
SCRAPOXY_REDIS_CONNECTION_POOL_SIZE = 20
SCHEDULER_FLUSH_ON_START = True  # 启动时清空队列

# 监控配置
STATS_CLASS = 'scrapy_redis.stats.RedisStatsCollector'
REDIS_STATS_KEY = '%(spider)s:stats'  # 统计数据键名

# 消息队列配置
REDIS_JOBS_KEY = '%(spider)s:jobs'  # 任务队列键名
REDIS_ITEMS_KEY = '%(spider)s:items'  # 结果队列键名

# scheduler.py - 调度器详解
from scrapy_redis.scheduler import Scheduler
from scrapy.utils.misc import load_object
import redis

class CustomScheduler(Scheduler):
    """
    自定义调度器
    """
    
    def __init__(self, server, persist=False, flush_on_start=False, queue_cls=None, 
                 dupefilter_cls=None, idle_persist=False, serializer=None, **kwargs):
        """
        初始化调度器
        """
        super().__init__(
            server=server,
            persist=persist,
            flush_on_start=flush_on_start,
            queue_cls=queue_cls,
            dupefilter_cls=dupefilter_cls,
            idle_persist=idle_persist,
            serializer=serializer,
            **kwargs
        )
        
        # 自定义初始化逻辑
        self.custom_stats = {}
        self.task_distribution_strategy = 'round_robin'
    
    def enqueue_request(self, request):
        """
        入队请求
        """
        if not request.dont_filter and self.df.request_seen(request):
            # 已经见过的请求，跳过
            return False
        
        if self.server.llen(self.queue_key) == 0:
            # 队列为空时的处理
            self._initialize_queue()
        
        # 添加请求到队列
        self.queue.push(request)
        return True
    
    def next_request(self):
        """
        获取下一个请求
        """
        request = self.queue.pop()
        if request:
            # 更新统计信息
            self._update_request_stats(request)
        return request
    
    def _update_request_stats(self, request):
        """
        更新请求统计
        """
        # 记录请求来源、优先级等信息
        pass
    
    def _initialize_queue(self):
        """
        初始化队列
        """
        # 预处理队列，如设置初始权重等
        pass

# dupefilter.py - 去重过滤器详解
from scrapy_redis.dupefilter import RFPDupeFilter
import hashlib
import time

class AdvancedDupeFilter(RFPDupeFilter):
    """
    高级去重过滤器
    """
    
    def __init__(self, server, key, debug=False, expire_time=86400*7):  # 7天过期
        """
        初始化去重过滤器
        """
        super().__init__(server, key, debug)
        self.expire_time = expire_time  # 过期时间
        self.fingerprint_cache = {}     # 指纹缓存
        self.cache_size_limit = 10000   # 缓存大小限制
    
    def request_seen(self, request):
        """
        检查请求是否已见过
        """
        fp = self.request_fingerprint(request)
        
        # 检查本地缓存
        if fp in self.fingerprint_cache:
            return True
        
        # 检查Redis集合
        if self.server.sismember(self.key, fp):
            # 缓存到本地，提高后续检查速度
            self._cache_fingerprint(fp)
            return True
        
        # 添加到Redis集合
        self.server.sadd(self.key, fp)
        
        # 设置过期时间
        if self.expire_time:
            self.server.expire(self.key, self.expire_time)
        
        # 缓存到本地
        self._cache_fingerprint(fp)
        return False
    
    def request_fingerprint(self, request):
        """
        生成请求指纹
        """
        # 使用更复杂的指纹生成算法
        fingerprint_str = f"{request.method}:{request.url}:{request.body.decode('utf-8', errors='ignore')}"
        return hashlib.sha256(fingerprint_str.encode()).hexdigest()
    
    def _cache_fingerprint(self, fp):
        """
        缓存指纹到本地
        """
        if len(self.fingerprint_cache) >= self.cache_size_limit:
            # 清理一半缓存
            keys_to_remove = list(self.fingerprint_cache.keys())[:len(self.fingerprint_cache)//2]
            for key in keys_to_remove:
                del self.fingerprint_cache[key]
        
        self.fingerprint_cache[fp] = time.time()
    
    def clear_cache(self):
        """
        清理缓存
        """
        self.fingerprint_cache.clear()

# queues.py - 请求队列详解
from scrapy_redis.queue import PriorityQueue, FifoQueue, LifoQueue
import pickle
import json

class OptimizedPriorityQueue(PriorityQueue):
    """
    优化的优先级队列
    """
    
    def __init__(self, server, spider, key, serializer=None):
        """
        初始化队列
        """
        super().__init__(server, spider, key, serializer)
        self.batch_size = 100  # 批量操作大小
        self.compression_enabled = True  # 启用压缩
    
    def _encode_request(self, request):
        """
        编码请求
        """
        obj = super()._encode_request(request)
        
        if self.compression_enabled:
            # 启用压缩以节省Redis空间
            import zlib
            return zlib.compress(pickle.dumps(obj))
        
        return pickle.dumps(obj)
    
    def _decode_request(self, encoded_request):
        """
        解码请求
        """
        if self.compression_enabled:
            import zlib
            obj = pickle.loads(zlib.decompress(encoded_request))
        else:
            obj = pickle.loads(encoded_request)
        
        return self.serializer.loads(obj)
    
    def push_batch(self, requests):
        """
        批量推送请求
        """
        pipe = self.server.pipeline()
        
        for request in requests:
            priority = -request.priority
            encoded_request = self._encode_request(request)
            pipe.zadd(self.key, {encoded_request: priority})
        
        pipe.execute()
    
    def pop_batch(self, batch_size=10):
        """
        批量弹出请求
        """
        # 按优先级获取多个请求
        items = self.server.zrange(self.key, 0, batch_size - 1)
        
        if items:
            pipe = self.server.pipeline()
            
            # 批量删除已获取的请求
            for item in items:
                pipe.zrem(self.key, item)
            
            pipe.execute()
            
            # 解码并返回请求
            return [self._decode_request(item) for item in items]
        
        return []

class AdaptiveQueue:
    """
    自适应队列 - 根据负载动态调整策略
    """
    
    def __init__(self, server, spider, base_key):
        self.server = server
        self.spider = spider
        self.base_key = base_key
        self.queues = {}
        self.load_balancer = LoadBalancer()
    
    def push(self, request):
        """
        智能推送请求到合适队列
        """
        target_queue = self._select_target_queue(request)
        return self.queues[target_queue].push(request)
    
    def pop(self):
        """
        从最优队列获取请求
        """
        selected_queue = self._select_optimal_queue()
        return self.queues[selected_queue].pop()
    
    def _select_target_queue(self, request):
        """
        根据请求特征选择目标队列
        """
        # 根据URL域名、请求类型等因素选择队列
        domain = request.url.split('/')[2]
        return f"{self.base_key}:{domain}"
    
    def _select_optimal_queue(self):
        """
        选择最优队列
        """
        # 根据队列长度、处理速度等因素选择
        queue_lengths = {
            key: self.server.llen(key) 
            for key in self.queues.keys()
        }
        
        # 选择长度最小的队列
        return min(queue_lengths, key=queue_lengths.get)

# distributed_spider.py - 分布式爬虫实现
from scrapy_redis.spiders import RedisSpider
from scrapy.http import Request
import json
import time

class DistributedSpider(RedisSpider):
    """
    分布式爬虫基类
    """
    name = 'distributed_spider'
    
    # Redis键名 - 所有实例共享此队列
    redis_key = 'distributed_spider:start_urls'
    
    # 最大并发请求数
    custom_settings = {
        'CONCURRENT_REQUESTS': 32,
        'CONCURRENT_REQUESTS_PER_DOMAIN': 4,
        'DOWNLOAD_DELAY': 1,
        'RANDOMIZE_DOWNLOAD_DELAY': 0.5,
        'AUTOTHROTTLE_ENABLED': True,
        'AUTOTHROTTLE_START_DELAY': 1,
        'AUTOTHROTTLE_MAX_DELAY': 10,
        'AUTOTHROTTLE_TARGET_CONCURRENCY': 4.0,
    }
    
    def __init__(self, *args, **kwargs):
        """
        初始化分布式爬虫
        """
        super().__init__(*args, **kwargs)
        
        # 初始化统计信息
        self.stats = {
            'requests_sent': 0,
            'responses_received': 0,
            'items_scraped': 0,
            'errors': 0,
            'start_time': time.time()
        }
    
    def make_requests_from_url(self, url):
        """
        从URL创建请求
        """
        # 支持自定义请求参数
        return Request(
            url=url,
            callback=self.parse,
            errback=self.handle_error,
            meta={
                'download_timeout': 30,
                'dont_redirect': False,
                'handle_httpstatus_list': [301, 302, 404, 500]
            }
        )
    
    def parse(self, response):
        """
        解析响应
        """
        try:
            # 更新统计信息
            self.stats['responses_received'] += 1
            
            # 提取数据
            data = {
                'url': response.url,
                'status': response.status,
                'title': response.css('title::text').get(),
                'timestamp': time.time(),
                'spider_name': self.name,
                'node_id': self._get_node_id()
            }
            
            yield data
            
            # 提取链接并加入队列
            for link in response.css('a::attr(href)').getall():
                absolute_url = response.urljoin(link)
                yield Request(
                    url=absolute_url,
                    callback=self.parse,
                    meta={'depth': response.meta.get('depth', 0) + 1}
                )
                
        except Exception as e:
            self.logger.error(f"Parse error: {response.url}, Error: {str(e)}")
            self.stats['errors'] += 1
    
    def handle_error(self, failure):
        """
        处理请求错误
        """
        self.logger.error(f"Request failed: {failure.request.url}")
        self.logger.error(f"Reason: {failure.value}")
        self.stats['errors'] += 1
    
    def _get_node_id(self):
        """
        获取节点ID
        """
        import socket
        return socket.gethostname()
    
    def closed(self, reason):
        """
        爬虫关闭时的清理工作
        """
        self.logger.info(f"Spider closed. Reason: {reason}")
        self.logger.info(f"Final stats: {self.stats}")

class AdvancedDistributedSpider(DistributedSpider):
    """
    高级分布式爬虫
    """
    
    name = 'advanced_distributed_spider'
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        
        # 初始化高级功能
        self.rate_limiter = RateLimiter()
        self.domain_scheduler = DomainScheduler()
        self.task_manager = TaskManager()
    
    def parse(self, response):
        """
        高级解析逻辑
        """
        # 智能解析策略
        content_type = response.headers.get('content-type', b'').decode('utf-8')
        
        if 'json' in content_type:
            yield from self.parse_json_api(response)
        elif 'html' in content_type:
            yield from self.parse_html_page(response)
        else:
            yield from self.parse_generic_content(response)
    
    def parse_json_api(self, response):
        """
        解析JSON API响应
        """
        try:
            data = json.loads(response.text)
            
            # 递归提取URL
            urls = self._extract_urls_from_json(data)
            for url in urls:
                yield Request(url=url, callback=self.parse)
            
            # 提取业务数据
            yield {
                'type': 'api_data',
                'data': data,
                'url': response.url,
                'timestamp': time.time()
            }
            
        except json.JSONDecodeError:
            self.logger.error(f"Failed to decode JSON: {response.url}")
    
    def parse_html_page(self, response):
        """
        解析HTML页面
        """
        # 提取结构化数据
        structured_data = {
            'url': response.url,
            'title': response.css('title::text').get(),
            'meta_description': response.css('meta[name="description"]::attr(content)').get(),
            'links': response.css('a::attr(href)').getall(),
            'images': response.css('img::attr(src)').getall(),
            'headings': {
                'h1': response.css('h1::text').getall(),
                'h2': response.css('h2::text').getall(),
                'h3': response.css('h3::text').getall(),
            }
        }
        
        yield structured_data
    
    def _extract_urls_from_json(self, data, urls=None):
        """
        从JSON数据中提取URL
        """
        if urls is None:
            urls = []
        
        if isinstance(data, dict):
            for key, value in data.items():
                if key in ['url', 'link', 'href', 'src'] and isinstance(value, str):
                    if value.startswith(('http://', 'https://')):
                        urls.append(value)
                elif isinstance(value, (dict, list)):
                    self._extract_urls_from_json(value, urls)
        elif isinstance(data, list):
            for item in data:
                self._extract_urls_from_json(item, urls)
        
        return urls

class TaskBasedDistributedSpider(RedisSpider):
    """
    基于任务的分布式爬虫
    """
    
    name = 'task_based_spider'
    redis_key = 'task_spider:tasks'
    
    def parse(self, response):
        """
        解析任务
        """
        task_data = json.loads(response.url)  # 任务信息存储在URL中
        
        task_type = task_data.get('type')
        task_params = task_data.get('params', {})
        
        if task_type == 'crawl_page':
            yield from self._crawl_page(task_params)
        elif task_type == 'extract_data':
            yield from self._extract_data(task_params)
        elif task_type == 'api_call':
            yield from self._make_api_call(task_params)
    
    def _crawl_page(self, params):
        """
        爬取页面任务
        """
        url = params['url']
        selectors = params.get('selectors', {})
        
        # 发起页面请求
        yield Request(
            url=url,
            callback=self._parse_crawl_result,
            meta={'selectors': selectors}
        )
    
    def _parse_crawl_result(self, response):
        """
        解析爬取结果
        """
        selectors = response.meta['selectors']
        result = {}
        
        for field, selector in selectors.items():
            result[field] = response.css(selector).get()
        
        yield result

# task_dispatcher.py - 任务分发器
import redis
import json
import time
from typing import Dict, List, Any

class TaskDispatcher:
    """
    任务分发器
    """
    
    def __init__(self, redis_url='redis://localhost:6379', namespace='distributed_spider'):
        self.redis_client = redis.from_url(redis_url)
        self.namespace = namespace
        self.task_queue_key = f"{namespace}:task_queue"
        self.node_registry_key = f"{namespace}:nodes"
        self.task_status_key = f"{namespace}:task_status"
    
    def register_node(self, node_id: str, capabilities: Dict[str, Any]):
        """
        注册爬虫节点
        """
        node_info = {
            'id': node_id,
            'capabilities': capabilities,
            'registered_at': time.time(),
            'status': 'active'
        }
        
        self.redis_client.hset(
            self.node_registry_key, 
            node_id, 
            json.dumps(node_info)
        )
    
    def submit_task(self, task_data: Dict[str, Any], priority: int = 0):
        """
        提交任务
        """
        task_id = f"task_{int(time.time() * 1000000)}"
        
        task_info = {
            'id': task_id,
            'data': task_data,
            'priority': priority,
            'submitted_at': time.time(),
            'status': 'pending'
        }
        
        # 添加到任务队列
        self.redis_client.zadd(
            self.task_queue_key,
            {json.dumps(task_info): -priority}  # 负号实现降序排序
        )
        
        # 初始化任务状态
        self.redis_client.hset(
            self.task_status_key,
            task_id,
            json.dumps({'status': 'pending', 'submitted_at': time.time()})
        )
        
        return task_id
    
    def get_available_nodes(self) -> List[Dict[str, Any]]:
        """
        获取可用节点
        """
        nodes = []
        node_keys = self.redis_client.hkeys(self.node_registry_key)
        
        for node_key in node_keys:
            node_info = self.redis_client.hget(self.node_registry_key, node_key)
            if node_info:
                node_data = json.loads(node_info)
                if node_data.get('status') == 'active':
                    nodes.append(node_data)
        
        return nodes
    
    def dispatch_tasks(self, max_tasks_per_cycle: int = 100):
        """
        分发任务到节点
        """
        # 获取待处理任务
        pending_tasks = self.redis_client.zrange(
            self.task_queue_key, 
            0, 
            max_tasks_per_cycle - 1
        )
        
        available_nodes = self.get_available_nodes()
        
        if not pending_tasks or not available_nodes:
            return
        
        # 简单轮询分配
        for i, task_data in enumerate(pending_tasks):
            if i >= len(available_nodes):
                break
            
            task_info = json.loads(task_data)
            target_node = available_nodes[i % len(available_nodes)]
            
            # 将任务分发到节点专用队列
            node_queue_key = f"{self.namespace}:node:{target_node['id']}:queue"
            self.redis_client.lpush(node_queue_key, task_data)
            
            # 更新任务状态
            self.redis_client.hset(
                self.task_status_key,
                task_info['id'],
                json.dumps({
                    'status': 'dispatched',
                    'dispatched_to': target_node['id'],
                    'dispatched_at': time.time()
                })
            )
            
            # 从主队列移除任务
            self.redis_client.zrem(self.task_queue_key, task_data)

class LoadBalancer:
    """
    负载均衡器
    """
    
    def __init__(self, redis_client):
        self.redis_client = redis_client
        self.node_loads_key = 'distributed_spider:node_loads'
        self.task_distribution_key = 'distributed_spider:task_distribution'
    
    def get_optimal_node(self, task_type: str = 'default') -> str:
        """
        获取最优节点
        """
        # 获取所有节点负载
        node_loads = self.redis_client.hgetall(self.node_loads_key)
        
        if not node_loads:
            return 'default_node'
        
        # 找到负载最低的节点
        min_load = float('inf')
        optimal_node = None
        
        for node_id, load_str in node_loads.items():
            try:
                load = int(load_str.decode('utf-8'))
                if load < min_load:
                    min_load = load
                    optimal_node = node_id.decode('utf-8')
            except (ValueError, UnicodeDecodeError):
                continue
        
        return optimal_node or 'default_node'
    
    def update_node_load(self, node_id: str, increment: int = 1):
        """
        更新节点负载
        """
        self.redis_client.hincrby(self.node_loads_key, node_id, increment)
    
    def record_task_distribution(self, node_id: str, task_type: str):
        """
        记录任务分布
        """
        distribution_key = f"{self.task_distribution_key}:{task_type}"
        self.redis_client.hincrby(distribution_key, node_id, 1)

# 使用示例
if __name__ == "__main__":
    dispatcher = TaskDispatcher()
    
    # 注册节点
    dispatcher.register_node(
        'node_1', 
        {'cpu_cores': 8, 'memory_gb': 16, 'bandwidth_mbps': 100}
    )
    
    # 提交任务
    task_id = dispatcher.submit_task({
        'type': 'crawl',
        'url': 'https://example.com',
        'depth': 2
    }, priority=1)
    
    print(f"Submitted task: {task_id}")

# queue_strategy.py - 队列管理策略
from scrapy_redis.queue import PriorityQueue
import time
import hashlib

class MultiQueueStrategy:
    """
    多队列管理策略
    """
    
    def __init__(self, redis_client, base_namespace='distributed_spider'):
        self.redis_client = redis_client
        self.base_namespace = base_namespace
        self.queue_configs = {}
        self.routing_rules = []
    
    def add_queue_config(self, queue_name: str, config: dict):
        """
        添加队列配置
        """
        self.queue_configs[queue_name] = config
    
    def add_routing_rule(self, rule_func, target_queue: str):
        """
        添加路由规则
        """
        self.routing_rules.append((rule_func, target_queue))
    
    def route_request(self, request):
        """
        路由请求到合适队列
        """
        for rule_func, target_queue in self.routing_rules:
            if rule_func(request):
                return target_queue
        
        # 默认队列
        return 'default_queue'
    
    def prioritize_request(self, request, base_priority: int = 0):
        """
        为请求计算优先级
        """
        priority = base_priority
        
        # 根据URL特征调整优先级
        url = request.url
        if 'important' in url:
            priority += 100
        elif 'urgent' in url:
            priority += 50
        
        # 根据域名频率限制调整
        domain = url.split('/')[2]
        recent_requests = self.redis_client.get(f"recent_requests:{domain}")
        if recent_requests and int(recent_requests) > 10:
            priority -= 20  # 降低频繁域名的优先级
        
        # 根据请求深度调整
        depth = request.meta.get('depth', 0)
        if depth > 5:
            priority -= depth * 5  # 深度越大优先级越低
        
        return priority

class PriorityAwareQueue(PriorityQueue):
    """
    支持优先级感知的队列
    """
    
    def __init__(self, server, spider, key, serializer=None):
        super().__init__(server, spider, key, serializer)
        self.priority_stats_key = f"{key}:priority_stats"
        self.stats_window = 3600  # 1小时统计窗口
    
    def push(self, request):
        """
        推送请求时更新优先级统计
        """
        priority = -request.priority
        result = super().push(request)
        
        # 更新优先级统计
        self._update_priority_stats(priority)
        return result
    
    def _update_priority_stats(self, priority: int):
        """
        更新优先级统计
        """
        current_time = int(time.time())
        hour_bucket = current_time - (current_time % self.stats_window)
        
        # 记录优先级分布
        self.redis_client.hincrby(
            f"{self.priority_stats_key}:{hour_bucket}",
            f"priority_{priority}",
            1
        )
    
    def get_queue_statistics(self):
        """
        获取队列统计信息
        """
        current_time = int(time.time())
        hour_bucket = current_time - (current_time % self.stats_window)
        
        stats = self.redis_client.hgetall(f"{self.priority_stats_key}:{hour_bucket}")
        return {
            key.decode('utf-8'): int(value.decode('utf-8'))
            for key, value in stats.items()
        }

class AdaptiveQueueManager:
    """
    自适应队列管理器
    """
    
    def __init__(self, redis_client):
        self.redis_client = redis_client
        self.queue_performance_key = 'queue_performance'
        self.adaptation_threshold = 0.8  # 性能阈值
    
    def evaluate_queue_performance(self, queue_name: str) -> float:
        """
        评估队列性能
        """
        # 获取队列性能指标
        processed_count = self.redis_client.get(f"queue:{queue_name}:processed")
        error_count = self.redis_client.get(f"queue:{queue_name}:errors")
        timeout_count = self.redis_client.get(f"queue:{queue_name}:timeouts")
        
        processed = int(processed_count) if processed_count else 0
        errors = int(error_count) if error_count else 0
        timeouts = int(timeout_count) if timeout_count else 0
        
        total = processed + errors + timeouts
        if total == 0:
            return 1.0  # 默认性能
        
        success_rate = processed / total if total > 0 else 0
        error_rate = (errors + timeouts) / total if total > 0 else 1
        
        # 综合评估
        performance = success_rate * 0.7 - error_rate * 0.3
        return max(0, min(1, performance))  # 归一化到[0,1]
    
    def adapt_queue_configuration(self, queue_name: str):
        """
        自适应调整队列配置
        """
        performance = self.evaluate_queue_performance(queue_name)
        
        if performance < self.adaptation_threshold:
            # 性能不佳，降低负载
            current_concurrency = self.redis_client.get(f"queue:{queue_name}:concurrency")
            if current_concurrency:
                new_concurrency = max(1, int(current_concurrency) // 2)
                self.redis_client.set(f"queue:{queue_name}:concurrency", new_concurrency)
        else:
            # 性能良好，可以增加负载
            current_concurrency = self.redis_client.get(f"queue:{queue_name}:concurrency")
            if current_concurrency:
                new_concurrency = min(100, int(current_concurrency) * 2)
                self.redis_client.set(f"queue:{queue_name}:concurrency", new_concurrency)

class QueuePartitioner:
    """
    队列分区器
    """
    
    def __init__(self, num_partitions: int = 10):
        self.num_partitions = num_partitions
    
    def partition_by_domain(self, url: str) -> int:
        """
        按域名分区
        """
        domain = url.split('/')[2]
        hash_value = int(hashlib.md5(domain.encode()).hexdigest(), 16)
        return hash_value % self.num_partitions
    
    def partition_by_content_type(self, content_type: str) -> int:
        """
        按内容类型分区
        """
        type_hash = int(hashlib.md5(content_type.encode()).hexdigest(), 16)
        return type_hash % self.num_partitions
    
    def partition_by_priority(self, priority: int) -> int:
        """
        按优先级分区
        """
        return abs(priority) % self.num_partitions

# deduplication_analysis.py - 去重机制分析
import hashlib
import mmh3
import time
from bitarray import bitarray
import redis

class BloomFilter:
    """
    布隆过滤器实现
    """
    
    def __init__(self, capacity: int, error_rate: float = 0.01):
        """
        初始化布隆过滤器
        """
        self.capacity = capacity
        self.error_rate = error_rate
        
        # 计算位数组大小和哈希函数数量
        self.bit_array_size = self._get_bit_array_size()
        self.hash_count = self._get_hash_count()
        
        self.bit_array = bitarray(self.bit_array_size)
        self.bit_array.setall(0)
    
    def _get_bit_array_size(self) -> int:
        """
        计算位数组大小
        """
        import math
        m = -(self.capacity * math.log(self.error_rate)) / (math.log(2) ** 2)
        return int(m)
    
    def _get_hash_count(self) -> int:
        """
        计算哈希函数数量
        """
        import math
        k = (self.bit_array_size / self.capacity) * math.log(2)
        return int(k)
    
    def _hash(self, item: str, seed: int) -> int:
        """
        生成哈希值
        """
        return mmh3.hash(item, seed) % self.bit_array_size
    
    def add(self, item: str):
        """
        添加元素
        """
        for i in range(self.hash_count):
            index = self._hash(item, i)
            self.bit_array[index] = 1
    
    def contains(self, item: str) -> bool:
        """
        检查元素是否存在
        """
        for i in range(self.hash_count):
            index = self._hash(item, i)
            if not self.bit_array[index]:
                return False
        return True

class RedisBloomFilter:
    """
    基于Redis的布隆过滤器
    """
    
    def __init__(self, redis_client, key_prefix: str, capacity: int = 1000000, error_rate: float = 0.01):
        self.redis_client = redis_client
        self.key_prefix = key_prefix
        self.capacity = capacity
        self.error_rate = error_rate
        
        # 计算参数
        import math
        self.bit_array_size = int(-(capacity * math.log(error_rate)) / (math.log(2) ** 2))
        self.hash_count = int((self.bit_array_size / capacity) * math.log(2))
        
        # Redis位图键
        self.bitmap_key = f"{key_prefix}:bloom_bitmap"
    
    def _get_hashes(self, item: str) -> list:
        """
        获取多个哈希值
        """
        hashes = []
        for i in range(self.hash_count):
            hash_val = mmh3.hash(item, i) % self.bit_array_size
            hashes.append(hash_val)
        return hashes
    
    def add(self, item: str):
        """
        添加元素到布隆过滤器
        """
        hashes = self._get_hashes(item)
        pipe = self.redis_client.pipeline()
        
        for hash_val in hashes:
            pipe.setbit(self.bitmap_key, hash_val, 1)
        
        pipe.execute()
    
    def contains(self, item: str) -> bool:
        """
        检查元素是否存在
        """
        hashes = self._get_hashes(item)
        
        for hash_val in hashes:
            if not self.redis_client.getbit(self.bitmap_key, hash_val):
                return False
        
        return True
    
    def clear(self):
        """
        清空布隆过滤器
        """
        self.redis_client.delete(self.bitmap_key)

class AdvancedDupeFilter:
    """
    高级去重过滤器
    """
    
    def __init__(self, redis_client, namespace: str = 'distributed_spider'):
        self.redis_client = redis_client
        self.namespace = namespace
        
        # 多层去重机制
        self.local_cache = {}  # 本地缓存
        self.redis_set_key = f"{namespace}:dupefilter:set"  # Redis集合
        self.bloom_filter = RedisBloomFilter(
            redis_client, 
            f"{namespace}:dupefilter:bloom"
        )
        
        # 性能统计
        self.stats = {
            'local_hits': 0,
            'redis_hits': 0,
            'bloom_hits': 0,
            'misses': 0,
            'false_positives': 0
        }
    
    def request_seen(self, request) -> bool:
        """
        检查请求是否已见过
        """
        fingerprint = self._get_fingerprint(request)
        
        # 1. 检查本地缓存（最快）
        if fingerprint in self.local_cache:
            self.stats['local_hits'] += 1
            return True
        
        # 2. 检查布隆过滤器（可能存在误判）
        if self.bloom_filter.contains(fingerprint):
            # 布隆过滤器可能误判，需要检查Redis集合确认
            if self.redis_client.sismember(self.redis_set_key, fingerprint):
                self.stats['redis_hits'] += 1
                self._cache_fingerprint(fingerprint)  # 缓存到本地
                return True
            else:
                # 布隆过滤器误判
                self.stats['false_positives'] += 1
        
        # 3. 检查Redis集合
        if self.redis_client.sismember(self.redis_set_key, fingerprint):
            self.stats['redis_hits'] += 1
            self._cache_fingerprint(fingerprint)
            return True
        
        # 4. 添加到所有层级
        pipe = self.redis_client.pipeline()
        pipe.sadd(self.redis_set_key, fingerprint)
        pipe.execute()
        
        self.bloom_filter.add(fingerprint)
        self._cache_fingerprint(fingerprint)
        
        self.stats['misses'] += 1
        return False
    
    def _get_fingerprint(self, request) -> str:
        """
        生成请求指纹
        """
        # 使用URL、方法、参数等生成唯一指纹
        fingerprint_str = f"{request.method}:{request.url}:{request.body.decode('utf-8', errors='ignore')}"
        return hashlib.sha256(fingerprint_str.encode()).hexdigest()
    
    def _cache_fingerprint(self, fingerprint: str):
        """
        缓存指纹到本地
        """
        # 限制缓存大小
        if len(self.local_cache) > 10000:
            # 移除最早的一半缓存
            keys_to_remove = list(self.local_cache.keys())[:5000]
            for key in keys_to_remove:
                del self.local_cache[key]
        
        self.local_cache[fingerprint] = time.time()
    
    def get_stats(self) -> dict:
        """
        获取统计信息
        """
        total_checks = sum(self.stats.values())
        hit_rate = (self.stats['local_hits'] + self.stats['redis_hits']) / total_checks if total_checks > 0 else 0
        
        return {
            **self.stats,
            'hit_rate': hit_rate,
            'cache_size': len(self.local_cache)
        }

class DuplicateAnalysis:
    """
    重复数据分析师
    """
    
    def __init__(self, redis_client):
        self.redis_client = redis_client
        self.analysis_key = 'duplicate_analysis'
    
    def analyze_duplication_patterns(self, days: int = 7) -> dict:
        """
        分析重复模式
        """
        patterns = {
            'daily_duplication_rate': {},
            'top_duplicate_domains': {},
            'duplication_by_hour': {},
            'similarity_clusters': []
        }
        
        # 分析每日重复率
        for day in range(days):
            date_key = f"{self.analysis_key}:day_{day}"
            daily_stats = self.redis_client.hgetall(date_key)
            
            if daily_stats:
                processed = int(daily_stats.get(b'processed', 0))
                duplicates = int(daily_stats.get(b'duplicates', 0))
                
                if processed > 0:
                    duplication_rate = duplicates / processed
                    patterns['daily_duplication_rate'][day] = duplication_rate
        
        # 分析域名重复情况
        domain_stats = self.redis_client.hgetall(f"{self.analysis_key}:domains")
        for domain, count in domain_stats.items():
            patterns['top_duplicate_domains'][domain.decode()] = int(count)
        
        # 按小时分析
        hourly_stats = self.redis_client.hgetall(f"{self.analysis_key}:hours")
        for hour, count in hourly_stats.items():
            patterns['duplication_by_hour'][int(hour)] = int(count)
        
        return patterns
    
    def identify_similar_requests(self, threshold: float = 0.8) -> list:
        """
        识别相似请求
        """
        # 获取最近的请求样本
        recent_requests = self.redis_client.lrange(
            f"{self.analysis_key}:recent_requests", 0, 1000
        )
        
        similar_pairs = []
        
        for i, req1 in enumerate(recent_requests):
            for j, req2 in enumerate(recent_requests[i+1:], i+1):
                similarity = self._calculate_similarity(req1, req2)
                if similarity > threshold:
                    similar_pairs.append({
                        'request1': req1.decode(),
                        'request2': req2.decode(),
                        'similarity': similarity
                    })
        
        return sorted(similar_pairs, key=lambda x: x['similarity'], reverse=True)
    
    def _calculate_similarity(self, req1: bytes, req2: bytes) -> float:
        """
        计算请求相似度
        """
        str1, str2 = req1.decode(), req2.decode()
        
        # 简单的余弦相似度计算
        set1, set2 = set(str1), set(str2)
        intersection = len(set1.intersection(set2))
        union = len(set1.union(set2))
        
        return intersection / union if union > 0 else 0

# load_balancing.py - 负载均衡策略
import time
import random
from typing import List, Dict, Any
import heapq

class LoadBalancer:
    """
    负载均衡器
    """
    
    def __init__(self, nodes: List[Dict[str, Any]]):
        self.nodes = nodes
        self.node_stats = {node['id']: {'load': 0, 'success_rate': 1.0, 'response_time': 0.1} 
                          for node in nodes}
        self.algorithm = 'weighted_round_robin'  # 默认算法
    
    def select_node(self, request_info: Dict[str, Any] = None) -> str:
        """
        选择最优节点
        """
        if self.algorithm == 'random':
            return self._random_selection()
        elif self.algorithm == 'round_robin':
            return self._round_robin_selection()
        elif self.algorithm == 'weighted_round_robin':
            return self._weighted_round_robin_selection()
        elif self.algorithm == 'least_connections':
            return self._least_connections_selection()
        elif self.algorithm == 'response_time':
            return self._response_time_selection()
        elif self.algorithm == 'adaptive':
            return self._adaptive_selection(request_info)
        else:
            return self._round_robin_selection()
    
    def _random_selection(self) -> str:
        """
        随机选择
        """
        return random.choice(self.nodes)['id']
    
    def _round_robin_selection(self) -> str:
        """
        轮询选择
        """
        # 简单实现，实际中需要维护轮询状态
        active_nodes = [node for node in self.nodes if node.get('status') == 'active']
        if not active_nodes:
            return self.nodes[0]['id'] if self.nodes else None
        return active_nodes[hash(time.time()) % len(active_nodes)]['id']
    
    def _weighted_round_robin_selection(self) -> str:
        """
        加权轮询选择
        """
        # 根据节点能力和当前负载计算权重
        weighted_nodes = []
        
        for node in self.nodes:
            if node.get('status') != 'active':
                continue
            
            # 计算综合权重
            capability_weight = node.get('cpu_cores', 1) * node.get('memory_gb', 1)
            load_factor = 1.0 / (1.0 + self.node_stats[node['id']]['load'])
            success_weight = self.node_stats[node['id']]['success_rate']
            
            weight = capability_weight * load_factor * success_weight
            weighted_nodes.append((weight, node['id']))
        
        if not weighted_nodes:
            return self.nodes[0]['id'] if self.nodes else None
        
        # 根据权重随机选择
        total_weight = sum(weight for weight, _ in weighted_nodes)
        rand_num = random.uniform(0, total_weight)
        
        cumulative_weight = 0
        for weight, node_id in weighted_nodes:
            cumulative_weight += weight
            if rand_num <= cumulative_weight:
                return node_id
        
        return weighted_nodes[-1][1]
    
    def _least_connections_selection(self) -> str:
        """
        最少连接数选择
        """
        active_nodes = [node for node in self.nodes if node.get('status') == 'active']
        if not active_nodes:
            return self.nodes[0]['id'] if self.nodes else None
        
        min_load_node = min(active_nodes, 
                           key=lambda n: self.node_stats[n['id']]['load'])
        return min_load_node['id']
    
    def _response_time_selection(self) -> str:
        """
        响应时间选择
        """
        active_nodes = [node for node in self.nodes if node.get('status') == 'active']
        if not active_nodes:
            return self.nodes[0]['id'] if self.nodes else None
        
        # 选择平均响应时间最短的节点
        best_node = min(active_nodes,
                       key=lambda n: self.node_stats[n['id']]['response_time'])
        return best_node['id']
    
    def _adaptive_selection(self, request_info: Dict[str, Any]) -> str:
        """
        自适应选择
        """
        # 根据请求特征和节点能力进行智能匹配
        if request_info:
            # 如果请求包含特定要求，优先匹配有能力的节点
            required_resources = request_info.get('resources', {})
            
            for node in self.nodes:
                if node.get('status') != 'active':
                    continue
                
                # 检查资源匹配度
                resource_match = self._check_resource_match(node, required_resources)
                if resource_match:
                    return node['id']
        
        # 回退到加权轮询
        return self._weighted_round_robin_selection()
    
    def _check_resource_match(self, node: dict, required_resources: dict) -> bool:
        """
        检查资源匹配度
        """
        for resource, required_value in required_resources.items():
            node_value = node.get(resource, 0)
            if node_value < required_value:
                return False
        return True
    
    def update_node_stats(self, node_id: str, load_change: int = 0, 
                         success: bool = True, response_time: float = 0.0):
        """
        更新节点统计信息
        """
        if node_id not in self.node_stats:
            return
        
        stats = self.node_stats[node_id]
        stats['load'] += load_change
        
        # 更新成功率（指数移动平均）
        alpha = 0.1  # 平滑因子
        current_success_rate = 1.0 if success else 0.0
        stats['success_rate'] = (alpha * current_success_rate + 
                                (1 - alpha) * stats['success_rate'])
        
        # 更新响应时间（指数移动平均）
        current_response_time = response_time or stats['response_time']
        stats['response_time'] = (alpha * current_response_time + 
                                 (1 - alpha) * stats['response_time'])

class ConsistentHashBalancer:
    """
    一致性哈希负载均衡器
    """
    
    def __init__(self, nodes: List[str], virtual_nodes: int = 150):
        self.nodes = nodes
        self.virtual_nodes = virtual_nodes
        self.ring = {}  # 哈希环
        self.sorted_keys = []  # 排序的哈希值
        
        self._initialize_ring()
    
    def _initialize_ring(self):
        """
        初始化哈希环
        """
        for node in self.nodes:
            for i in range(self.virtual_nodes):
                virtual_key = f"{node}:{i}"
                hash_key = hash(virtual_key) % (2**32)
                self.ring[hash_key] = node
                self.sorted_keys.append(hash_key)
        
        self.sorted_keys.sort()
    
    def get_node(self, key: str) -> str:
        """
        根据键获取节点
        """
        if not self.ring:
            return None
        
        hash_key = hash(key) % (2**32)
        
        # 在排序的键中找到第一个大于等于hash_key的位置
        for ring_key in self.sorted_keys:
            if hash_key <= ring_key:
                return self.ring[ring_key]
        
        # 如果没找到，返回第一个节点（环形结构）
        return self.ring[self.sorted_keys[0]]

class TaskAffinityBalancer:
    """
    任务亲和性负载均衡器
    """
    
    def __init__(self, nodes: List[Dict[str, Any]]):
        self.nodes = nodes
        self.task_affinities = {}  # 任务-节点亲和性映射
        self.node_loads = {node['id']: 0 for node in nodes}
        self.max_affinity_duration = 3600  # 亲和性持续时间（秒）
    
    def assign_task(self, task_id: str, task_attributes: Dict[str, Any]) -> str:
        """
        分配任务到节点
        """
        # 检查是否有亲和性节点
        affinity_node = self._get_affinity_node(task_id)
        if affinity_node:
            return affinity_node
        
        # 根据任务属性选择最优节点
        optimal_node = self._select_optimal_node(task_attributes)
        
        # 设置亲和性（如果适用）
        if self._should_create_affinity(task_attributes):
            self._create_affinity(task_id, optimal_node)
        
        # 更新节点负载
        self.node_loads[optimal_node] += 1
        
        return optimal_node
    
    def _get_affinity_node(self, task_id: str) -> str:
        """
        获取亲和性节点
        """
        if task_id in self.task_affinities:
            node_id, timestamp = self.task_affinities[task_id]
            if time.time() - timestamp < self.max_affinity_duration:
                return node_id
            else:
                # 亲和性过期，删除
                del self.task_affinities[task_id]
        
        return None
    
    def _select_optimal_node(self, task_attributes: Dict[str, Any]) -> str:
        """
        选择最优节点
        """
        # 根据任务属性和节点能力进行匹配
        domain = task_attributes.get('domain', '')
        content_type = task_attributes.get('content_type', 'html')
        
        # 优先选择处理过相同域名的节点
        for node_id in self.node_loads.keys():
            if self._has_domain_experience(node_id, domain):
                if self.node_loads[node_id] < max(self.node_loads.values()) * 0.8:
                    return node_id
        
        # 使用加权负载均衡选择
        min_load = min(self.node_loads.values())
        candidates = [node_id for node_id, load in self.node_loads.items() 
                     if load == min_load]
        
        return random.choice(candidates) if candidates else list(self.node_loads.keys())[0]
    
    def _has_domain_experience(self, node_id: str, domain: str) -> bool:
        """
        检查节点是否有域名处理经验
        """
        # 这里可以实现具体的域名经验检查逻辑
        experience_key = f"node_experience:{node_id}:{domain}"
        # 实际实现中可能需要查询Redis或其他存储
        return False
    
    def _should_create_affinity(self, task_attributes: Dict[str, Any]) -> bool:
        """
        是否应该创建亲和性
        """
        # 对于需要保持会话或状态的任务创建亲和性
        return task_attributes.get('requires_session', False)
    
    def _create_affinity(self, task_id: str, node_id: str):
        """
        创建任务-节点亲和性
        """
        self.task_affinities[task_id] = (node_id, time.time())
    
    def release_task(self, task_id: str, node_id: str):
        """
        释放任务
        """
        self.node_loads[node_id] -= 1
        
        # 删除亲和性映射
        if task_id in self.task_affinities:
            del self.task_affinities[task_id]

class LoadBalancerManager:
    """
    负载均衡管理器
    """
    
    def __init__(self):
        self.balancers = {}
        self.performance_monitor = PerformanceMonitor()
    
    def register_balancer(self, name: str, balancer):
        """
        注册负载均衡器
        """
        self.balancers[name] = balancer
    
    def get_balancer(self, name: str):
        """
        获取负载均衡器
        """
        return self.balancers.get(name)
    
    def switch_algorithm(self, name: str, algorithm: str):
        """
        切换负载均衡算法
        """
        if name in self.balancers:
            balancer = self.balancers[name]
            if hasattr(balancer, 'algorithm'):
                balancer.algorithm = algorithm
    
    def get_optimal_balancer(self, traffic_pattern: str) -> str:
        """
        根据流量模式选择最优负载均衡器
        """
        performance_stats = self.performance_monitor.get_stats()
        
        if traffic_pattern == 'uniform':
            return 'round_robin'
        elif traffic_pattern == 'bursty':
            return 'least_connections'
        elif traffic_pattern == 'sticky':
            return 'consistent_hash'
        else:
            return 'adaptive'

class PerformanceMonitor:
    """
    性能监控器
    """
    
    def __init__(self):
        self.metrics = {}
        self.history = {}
    
    def record_metric(self, balancer_name: str, node_id: str, 
                     response_time: float, success: bool):
        """
        记录性能指标
        """
        key = f"{balancer_name}:{node_id}"
        
        if key not in self.metrics:
            self.metrics[key] = {
                'response_times': [],
                'success_rates': [],
                'throughput': 0
            }
        
        metrics = self.metrics[key]
        metrics['response_times'].append(response_time)
        metrics['success_rates'].append(1 if success else 0)
        metrics['throughput'] += 1
    
    def get_stats(self) -> Dict[str, Any]:
        """
        获取统计信息
        """
        stats = {}
        
        for key, metrics in self.metrics.items():
            response_times = metrics['response_times']
            success_rates = metrics['success_rates']
            
            if response_times:
                avg_response_time = sum(response_times) / len(response_times)
                success_rate = sum(success_rates) / len(success_rates) if success_rates else 0
                
                stats[key] = {
                    'avg_response_time': avg_response_time,
                    'success_rate': success_rate,
                    'throughput': metrics['throughput']
                }
        
        return stats