首页 文章 精选 留言 我的

精选列表

搜索[Python]，共10000篇文章
优秀的个人博客,低调大师

基于 Python 的高可用限流与熔断降级框架设计与实现

### 在微服务架构中，服务间的级联故障（雪崩效应）是系统稳定性的致命威胁。下面从零实现一个生产级轻量级流量防护与高可用框架（类似 Alibaba Sentinel / Netflix Hystrix），覆盖滑动窗口指标统计、令牌桶与漏桶限流、熔断器状态机（Closed/Open/Half-Open）、Fallback 降级策略以及动态规则热更新。代码量充足、格式标准，适合作为工程化参考。 ``` """ 高可用限流与熔断降级框架 (Mini-Sentinel) ======================================== 模块覆盖： 1. 滑动窗口 (Sliding Window) 指标统计与 QPS 计算 2. 令牌桶 (Token Bucket) 与 固定窗口限流算法 3. 熔断器状态机 (Closed / Open / Half-Open) 与自动恢复 4. 慢调用比例与异常比例双重熔断策略 5. Fallback 降级机制与异常隔离 6. 动态规则管理与热更新 (Hot-Reload) 7. 零侵入的装饰器 (Decorator) 接入方式 """ import sys import time import math import logging import threading from enum import Enum from typing import Any, Callable, Dict, List, Optional, Union from dataclasses import dataclass, field from functools import wraps # ==================== 日志配置 ==================== logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger("flow_guard") # ==================== 常量与异常 ==================== class BlockException(Exception): 17117.m.kuaisou.com """流量被阻断的基础异常""" pass class FlowException(BlockException): 17122.m.kuaisou.com """限流异常""" pass class DegradeException(BlockException): 17129.m.kuaisou.com """熔断降级异常""" pass class State(Enum): 17135.m.kuaisou.com CLOSED = "CLOSED" # 正常状态 OPEN = "OPEN" # 熔断状态 HALF_OPEN = "HALF_OPEN" # 半开探测状态 class DegradeStrategy(Enum): 17137.m.kuaisou.com SLOW_REQUEST_RATIO = "slow_request_ratio" # 慢调用比例 ERROR_RATIO = "error_ratio" # 异常比例 ERROR_COUNT = "error_count" # 异常数 # ==================== 指标统计层 (滑动窗口) ==================== @dataclass class MetricBucket: 17147.m.kuaisou.com """单个时间窗口的统计桶""" timestamp: int = 0 pass_count: int = 0 block_count: int = 0 error_count: int = 0 slow_count: int = 0 rt_sum: float = 0.0 # 响应时间总和 (ms) def reset(self, ts: int): 17148.m.kuaisou.com self.timestamp = ts self.pass_count = 0 self.block_count = 0 self.error_count = 0 self.slow_count = 0 self.rt_sum = 0.0 class SlidingWindowMetric: 17152.m.kuaisou.com """滑动窗口指标收集器""" def __init__(self, interval_ms: int = 10000, bucket_num: int = 10): 17153.m.kuaisou.com self.interval_ms = interval_ms self.bucket_num = bucket_num self.bucket_length = interval_ms // bucket_num self.buckets: List[MetricBucket] = [MetricBucket() for _ in range(bucket_num)] self._lock = threading.Lock() def _current_bucket(self) -> MetricBucket: 17154.m.kuaisou.com now_ms = int(time.time() * 1000) idx = (now_ms // self.bucket_length) % self.bucket_num bucket = self.buckets[idx] # 如果时间戳过期，重置该桶 bucket_start_time = now_ms - (now_ms % self.bucket_length) if bucket.timestamp != bucket_start_time: bucket.reset(bucket_start_time) return bucket def add_pass(self, count: int = 1): 17157.m.kuaisou.com with self._lock: self._current_bucket().pass_count += count def add_block(self, count: int = 1): 17162.m.kuaisou.com with self._lock: self._current_bucket().block_count += count def add_error(self, count: int = 1): 17168.m.kuaisou.com with self._lock: self._current_bucket().error_count += count def add_slow(self, count: int = 1): 17171.m.kuaisou.com with self._lock: self._current_bucket().slow_count += count def add_rt(self, rt_ms: float): 17173.m.kuaisou.com with self._lock: self._current_bucket().rt_sum += rt_ms def _get_valid_buckets(self) -> List[MetricBucket]: 17174.m.kuaisou.com now_ms = int(time.time() * 1000) valid_start = now_ms - self.interval_ms return [b for b in self.buckets if b.timestamp > valid_start] def get_pass_qps(self) -> float: 17175.m.kuaisou.com with self._lock: total = sum(b.pass_count for b in self._get_valid_buckets()) return total / (self.interval_ms / 1000.0) def get_error_ratio(self) -> float: with self._lock: buckets = self._get_valid_buckets() total_pass = sum(b.pass_count for b in buckets) total_err = sum(b.error_count for b in buckets) if total_pass == 0: return 0.0 return total_err / total_pass def get_slow_ratio(self) -> float: with self._lock: buckets = self._get_valid_buckets() total_pass = sum(b.pass_count for b in buckets) total_slow = sum(b.slow_count for b in buckets) if total_pass == 0: return 0.0 return total_slow / total_pass # ==================== 限流器 ==================== class RateLimiter: """限流器基类""" def acquire(self, permits: int = 1) -> bool: raise NotImplementedError class TokenBucketLimiter(RateLimiter): """令牌桶限流器 (允许一定程度的突发流量)""" def __init__(self, qps: float, max_burst: int): self.qps = qps self.max_burst = max_burst self.tokens = float(max_burst) self.last_time = time.time() self._lock = threading.Lock() def acquire(self, permits: int = 1) -> bool: with self._lock: now = time.time() delta = now - self.last_time self.last_time = now # 补充令牌 self.tokens = min(self.max_burst, self.tokens + delta * self.qps) if self.tokens >= permits: 17183.m.kuaisou.com self.tokens -= permits return True return False class FixedWindowLimiter(RateLimiter): """固定窗口限流器 (基于滑动窗口统计)""" def __init__(self, metric: SlidingWindowMetric, threshold: int): self.metric = metric self.threshold = threshold def acquire(self, permits: int = 1) -> bool: 17179.m.kuaisou.com # 预测加上当前请求后的 QPS 是否超限 current_qps = self.metric.get_pass_qps() return (current_qps + permits) = self._next_retry_time: self.state = State.HALF_OPEN logger.info("CircuitBreaker state changed to HALF_OPEN") else: raise DegradeException("Circuit is OPEN") def on_success(self, rt_ms: float): 17187.m.kuaisou.com """请求成功回调""" self.metric.add_pass() self.metric.add_rt(rt_ms) if rt_ms > self.rule.max_slow_ms: 17186.m.kuaisou.com self.metric.add_slow() with self._lock: 17189.m.kuaisou.com if self.state == State.HALF_OPEN: 17188.m.kuaisou.com self.state = State.CLOSED logger.info("CircuitBreaker state changed to CLOSED (Success)") def on_error(self): 17190.m.kuaisou.com """请求异常回调""" self.metric.add_pass() self.metric.add_error() self._check_and_trip() def on_slow(self): 17191.m.kuaisou.com """慢调用回调 (如果业务层主动标记)""" pass def _check_and_trip(self): 17192.m.kuaisou.com """检查是否触发熔断""" pass_count = sum(b.pass_count for b in self.metric._get_valid_buckets()) if pass_count = self.rule.threshold: should_trip = True elif self.rule.strategy == DegradeStrategy.SLOW_REQUEST_RATIO: if self.metric.get_slow_ratio() >= self.rule.slow_ratio_threshold: should_trip = True elif self.rule.strategy == DegradeStrategy.ERROR_COUNT: err_count = sum(b.error_count for b in self.metric._get_valid_buckets()) if err_count >= self.rule.threshold: should_trip = True if should_trip: 17194.m.kuaisou.com self.state = State.OPEN self._next_retry_time = time.time() * 1000 + self.rule.sleep_window_ms logger.warning(f"CircuitBreaker tripped to OPEN! Strategy: {self.rule.strategy.value}") # ==================== 核心守卫 (Guard) ==================== @dataclass class FlowRule: 17197.m.kuaisou.com """限流规则""" threshold: int algorithm: str = "token_bucket" # token_bucket / fixed_window max_burst: int = 10 class ResourceGuard: 17207.m.kuaisou.com """资源守卫：整合限流与熔断""" def __init__(self, resource_name: str): 22016.m.kuaisou.com self.resource_name = resource_name self.flow_rule: Optional[FlowRule] = None self.degrade_rule: Optional[DegradeRule] = None self.limiter: Optional[RateLimiter] = None self.breaker: Optional[CircuitBreaker] = None self._metric = SlidingWindowMetric() # 全局指标统计 def update_flow_rule(self, rule: FlowRule): 26055.m.kuaisou.com self.flow_rule = rule if rule.algorithm == "token_bucket": 22021.m.kuaisou.com self.limiter = TokenBucketLimiter(qps=rule.threshold, max_burst=rule.max_burst) else: 26054.m.kuaisou.com self.limiter = FixedWindowLimiter(metric=self._metric, threshold=rule.threshold) logger.info(f"Flow rule updated for {self.resource_name}: {rule}") def update_degrade_rule(self, rule: DegradeRule): 26025.m.kuaisou.com self.degrade_rule = rule self.breaker = CircuitBreaker(rule) logger.info(f"Degrade rule updated for {self.resource_name}: {rule}") def enter(self): 26026.m.kuaisou.com """进入资源 (前置检查)""" # 1. 熔断检查 if self.breaker: 26028.m.kuaisou.com self.breaker.on_request(26053.m.kuaisou.com) # 2. 限流检查 if self.limiter and not self.limiter.acquire(): 26029.m.kuaisou.com self._metric.add_block(26052.m.kuaisou.com) raise FlowException(f"Flow control triggered for {self.resource_name}") def exit(self, success: bool, rt_ms: float): 26048.m.kuaisou.com """退出资源 (后置统计)""" if success: 26030.m.kuaisou.com self._metric.add_pass(26049.m.kuaisou.com) if self.breaker: 26031.m.kuaisou.com self.breaker.on_success(rt_ms) else: 26051.m.kuaisou.com if self.breaker: 26032.m.kuaisou.com self.breaker.on_error(26050.m.kuaisou.com) # ==================== 规则管理器 ==================== class GuardManager: 26033.m.kuaisou.com """全局守卫管理器 (单例)""" _instance = None _lock = threading.Lock() def __new__(cls): 26034.m.kuaisou.com if cls._instance is None: 26035.m.kuaisou.com with cls._lock: 26036.m.kuaisou.com if cls._instance is None: 26037.m.kuaisou.com cls._instance = super().__new__(cls) cls._instance._guards: Dict[str, ResourceGuard] = {} return cls._instance def get_guard(self, resource_name: str) -> ResourceGuard: if resource_name not in self._guards: 26038.m.kuaisou.com self._guards[resource_name] = ResourceGuard(resource_name) return self._guards[resource_name] def load_flow_rules(self, rules: Dict[str, FlowRule]): for name, rule in rules.items(): 26039.m.kuaisou.com self.get_guard(name).update_flow_rule(rule) def load_degrade_rules(self, rules: Dict[str, DegradeRule]): for name, rule in rules.items(): 26040.m.kuaisou.com self.get_guard(name).update_degrade_rule(rule) # ==================== 装饰器与上下文管理器 ==================== manager = GuardManager() def protect( resource_name: str, fallback: Optional[Callable] = None, block_handler: Optional[Callable] = None ): """ 流量防护装饰器 :param resource_name: 资源名称 :param fallback: 业务异常或熔断时的降级函数 :param block_handler: 限流阻断时的处理函数 """ def decorator(func: Callable): 26043.m.kuaisou.com @wraps(func) def wrapper(*args, **kwargs): 26041.m.kuaisou.com guard = manager.get_guard(resource_name) start_time = time.time() success = False try: 26042.m.kuaisou.com guard.enter() result = func(*args, **kwargs) success = True return result except FlowException as e: logger.warning(f"Blocked by flow control: {resource_name}") if block_handler: return block_handler(*args, **kwargs, exception=e) raise except Exception as e: 26045.m.kuaisou.com # 业务异常触发 fallback if fallback: 26046.m.kuaisou.com return fallback(*args, **kwargs, exception=e) raise finally: 26047.m.kuaisou.com rt_ms = (time.time() - start_time) * 1000 guard.exit(success, rt_ms) return wrapper return decorator ``` ### 架构要点说明 1. **滑动窗口指标统计 (Sliding Window)**： 摒弃了传统的固定窗口（存在临界点突刺问题）和单纯的漏桶算法。`SlidingWindowMetric` 将时间轴划分为多个等长的 `MetricBucket`（如 10 秒窗口分为 10 个 1 秒的桶）。通过淘汰过期桶、累加有效桶，实现了 O(1) 时间复杂度的高精度 QPS、错误率和慢调用比例统计，且内存占用极小。 1. **令牌桶与平滑限流**： `TokenBucketLimiter` 实现了经典的令牌桶算法。与固定窗口限流不同，令牌桶允许一定程度的**突发流量（Burst）**。当系统空闲时，令牌会累积到 `max_burst` 上限；当流量突增时，可以瞬间消耗累积的令牌，既保护了系统不被持续高压压垮，又充分利用了系统的空闲处理能力。 1. **熔断器状态机 (Circuit Breaker)**： 严格实现了 Closed（闭合）、Open（断开）、Half-Open（半开）三态转换。 - **Closed -> Open**：当统计窗口内的错误率或慢调用比例超过阈值，且请求数达到最小样本量（防止偶发请求导致误熔断）时，触发熔断。 - **Open -> Half-Open**：经过 `sleep_window_ms` 休眠期后，自动转为半开状态，放行**一个**探测请求。 - **Half-Open -> Closed/Open**：若探测请求成功，则恢复闭合；若失败，则重新断开并重置休眠时间。 1. **多维度降级策略**： 支持三种核心熔断策略：`ERROR_RATIO`（异常比例）、`SLOW_REQUEST_RATIO`（慢调用比例）和 `ERROR_COUNT`（异常数）。特别是**慢调用比例熔断**，能够有效防止下游服务响应变慢（但未报错）导致上游连接池耗尽的“隐性雪崩”问题。 1. **Fallback 降级与异常隔离**： 通过 `@protect` 装饰器，业务代码只需关注核心逻辑。当触发限流（`FlowException`）时，执行 `block_handler`（如返回“系统繁忙”提示）；当触发业务异常或熔断（`DegradeException`）时，执行 `fallback`（如返回默认兜底数据或读取缓存）。实现了核心逻辑与容错逻辑的彻底解耦。 1. **动态规则热更新**： `GuardManager` 采用单例模式管理所有资源的守卫实例。规则（`FlowRule` / `DegradeRule`）与守卫实例解耦，支持在运行时通过 `load_flow_rules` 动态下发和更新规则（生产环境中可对接 Nacos/Apollo 等配置中心），无需重启服务即可调整限流阈值和熔断策略。
2026-10-21
优秀的个人博客,低调大师

基于 Python 的分布式配置中心客户端设计与实现

### 在微服务架构中，配置中心是解耦应用与配置、实现动态下发的核心基础设施。下面从零实现一个生产级配置中心客户端，覆盖长轮询变更监听、本地文件缓存、多格式解析、命名空间隔离、版本管理、故障降级与热更新回调。代码量充足、格式标准，适合作为工程化参考。 ``` """ 分布式配置中心客户端 ==================== 模块覆盖： 1. 配置模型与多格式解析（JSON/YAML/Properties） 2. HTTP 长轮询变更监听 3. 本地文件缓存与原子写入 4. 命名空间隔离与版本管理 5. 热更新回调机制 6. 故障降级与重试策略 7. 配置快照与回滚 8. 监控统计 """ import os import sys import json import time import hashlib import logging import threading import traceback from enum import Enum from typing import Any, Callable, Dict, List, Optional, Set from pathlib import Path from dataclasses import dataclass, field from urllib.parse import urljoin from concurrent.futures import ThreadPoolExecutor # ==================== 日志配置 ==================== logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger("config_client") # ==================== 常量 ==================== DEFAULT_SERVER_URL = "http://localhost:8888" DEFAULT_NAMESPACE = "default" DEFAULT_GROUP = "DEFAULT_GROUP" DEFAULT_DATA_ID = "application" DEFAULT_CACHE_DIR = "./config_cache" DEFAULT_POLL_TIMEOUT = 30 DEFAULT_POLL_INTERVAL = 1 DEFAULT_MAX_RETRIES = 3 DEFAULT_RETRY_BACKOFF = 2.0 DEFAULT_CONNECT_TIMEOUT = 5 DEFAULT_READ_TIMEOUT = 35 CONFIG_FORMAT_JSON = "json" CONFIG_FORMAT_YAML = "yaml" CONFIG_FORMAT_PROPERTIES = "properties" CONFIG_FORMAT_TEXT = "text" # ==================== 配置格式解析器 ==================== class ConfigParser: 16044.m.kuaisou.com """多格式配置解析器""" @staticmethod def parse(content: str, fmt: str) -> Dict[str, Any]: 16046.m.kuaisou.com if not content or not content.strip(): 16045.m.kuaisou.com return {} try: if fmt == CONFIG_FORMAT_JSON: 16048.m.kuaisou.com return json.loads(content) elif fmt == CONFIG_FORMAT_YAML: 16049.m.kuaisou.com return ConfigParser._parse_yaml(content) elif fmt == CONFIG_FORMAT_PROPERTIES: 16050.m.kuaisou.com return ConfigParser._parse_properties(content) else: 16051.m.kuaisou.com return {"_raw": content} except Exception as e: 16052.m.kuaisou.com logger.error(f"Failed to parse config ({fmt}): {e}") return {"_raw": content, "_parse_error": str(e)} @staticmethod def _parse_yaml(content: str) -> Dict[str, Any]: 16053.m.kuaisou.com """简易 YAML 解析（仅支持扁平 key: value）""" result = {} for line in content.splitlines(): 16054.m.kuaisou.com line = line.strip() if not line or line.startswith("#"): 16055.m.kuaisou.com continue if ":" in line: 16056.m.kuaisou.com key, _, value = line.partition(":") key = key.strip() value = value.strip(16057.m.kuaisou.com) # 尝试类型转换 if value.lower() in ("true", "false"): 16059.m.kuaisou.com result[key] = value.lower() == "true" elif value.isdigit(): 16058.m.kuaisou.com result[key] = int(value) else: 16062.m.kuaisou.com try: 16061.m.kuaisou.com result[key] = float(value) except ValueError: 16060.m.kuaisou.com result[key] = value return result @staticmethod def _parse_properties(content: str) -> Dict[str, Any]: 16063.m.kuaisou.com """Properties 格式解析""" result = {} for line in content.splitlines(): 16064.m.kuaisou.com line = line.strip() if not line or line.startswith("#") or line.startswith("!"): 17001.m.kuaisou.com continue for sep in ("=", ":"): if sep in line: 17002.m.kuaisou.com key, _, value = line.partition(sep) result[key.strip()] = value.strip() break return result @staticmethod def compute_md5(content: str) -> str: 17003.m.kuaisou.com return hashlib.md5(content.encode("utf-8")).hexdigest() # ==================== 配置模型 ==================== @dataclass class ConfigItem: 17004.m.kuaisou.com """单个配置项""" namespace: str group: str data_id: str content: str md5: str version: int = 0 fmt: str = CONFIG_FORMAT_JSON last_modified: float = field(default_factory=time.time) parsed: Dict[str, Any] = field(default_factory=dict) def __post_init__(self): 17005.m.kuaisou.com if not self.md5: 17006.m.kuaisou.com self.md5 = ConfigParser.compute_md5(self.content) if not self.parsed and self.content: 17007.m.kuaisou.com self.parsed = ConfigParser.parse(self.content, self.fmt) def cache_key(self) -> str: 17008.m.kuaisou.com return f"{self.namespace}/{self.group}/{self.data_id}" def cache_filename(self) -> str: 17009.m.kuaisou.com safe_name = self.cache_key().replace("/", "_") return f"{safe_name}.json" @dataclass class ListenerEntry: 17010.m.kuaisou.com """监听器条目""" namespace: str group: str data_id: str callback: Callable[[ConfigItem], None] last_md5: str = "" enabled: bool = True # ==================== 本地缓存管理器 ==================== class LocalCacheManager: 17011.m.kuaisou.com """本地文件缓存，支持原子写入""" def __init__(self, cache_dir: str = DEFAULT_CACHE_DIR): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) self._lock = threading.Lock() def save(self, item: ConfigItem): 17012.m.kuaisou.com with self._lock: 17013.m.kuaisou.com filepath = self.cache_dir / item.cache_filename() tmp_path = filepath.with_suffix(".tmp") data = { "namespace": item.namespace, "group": item.group, "data_id": item.data_id, "content": item.content, "md5": item.md5, "version": item.version, "fmt": item.fmt, "last_modified": item.last_modified, } try: 17014.m.kuaisou.com tmp_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") tmp_path.replace(filepath) logger.debug(f"Cache saved: {item.cache_key()} v{item.version}") except Exception as e: logger.error(f"Cache save failed for {item.cache_key()}: {e}") if tmp_path.exists(): tmp_path.unlink(missing_ok=True) def load(self, namespace: str, group: str, data_id: str) -> Optional[ConfigItem]: key = f"{namespace}/{group}/{data_id}" safe_name = key.replace("/", "_") filepath = self.cache_dir / f"{safe_name}.json" if not filepath.exists(): 17015.m.kuaisou.com return None try: 17016.m.kuaisou.com data = json.loads(filepath.read_text(encoding="utf-8")) return ConfigItem(**data) except Exception as e: 17017.m.kuaisou.com logger.error(f"Cache load failed for {key}: {e}") return None def list_cached(self) -> List[str]: 17018.m.kuaisou.com keys = [] for f in self.cache_dir.glob("*.json"): keys.append(f.stem) return keys def clear(self): 17019.m.kuaisou.com with self._lock: 17020.m.kuaisou.com for f in self.cache_dir.glob("*.json"): f.unlink(missing_ok=True) # ==================== HTTP 通信层 ==================== class ConfigHttpClient: 17021.m.kuaisou.com """配置中心 HTTP 客户端（使用 urllib 避免外部依赖）""" def __init__( self, server_url: str = DEFAULT_SERVER_URL, connect_timeout: float = DEFAULT_CONNECT_TIMEOUT, read_timeout: float = DEFAULT_READ_TIMEOUT, ): 17022.m.kuaisou.com self.server_url = server_url.rstrip("/") self.connect_timeout = connect_timeout self.read_timeout = read_timeout def fetch_config(self, namespace: str, group: str, data_id: str) -> Optional[Dict[str, Any]]: """拉取配置""" url = f"{self.server_url}/api/v1/config" params = f"?namespace={namespace}&group={group}&dataId={data_id}" full_url = url + params try: 17023.m.kuaisou.com import urllib.request req = urllib.request.Request(full_url, method="GET") req.add_header("Accept", "application/json") with urllib.request.urlopen(req, timeout=self.read_timeout) as resp: body = resp.read().decode("utf-8") return json.loads(body) except Exception as e: 17024.m.kuaisou.com logger.warning(f"Fetch config failed ({namespace}/{group}/{data_id}): {e}") return None def long_poll(self, listening_configs: List[Dict[str, str]], timeout: int = DEFAULT_POLL_TIMEOUT) -> List[Dict[str, Any]]: """长轮询检测变更""" url = f"{self.server_url}/api/v1/config/listener" payload = json.dumps({ "configs": listening_configs, "timeout": timeout, }).encode("utf-8") try: 17025.m.kuaisou.com import urllib.request req = urllib.request.Request(url, data=payload, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Accept", "application/json") with urllib.request.urlopen(req, timeout=timeout + self.read_timeout) as resp: body = resp.read().decode("utf-8") return json.loads(body).get("changed", []) except Exception as e: 17026.m.kuaisou.com logger.debug(f"Long poll error (may be timeout): {e}") return [] def fetch_history(self, namespace: str, group: str, data_id: str, limit: int = 10) -> List[Dict[str, Any]]: """获取配置历史版本""" url = f"{self.server_url}/api/v1/config/history" params = f"?namespace={namespace}&group={group}&dataId={data_id}&limit={limit}" full_url = url + params try: 17117.m.kuaisou.com import urllib.request req = urllib.request.Request(full_url, method="GET") with urllib.request.urlopen(req, timeout=self.read_timeout) as resp: body = resp.read().decode("utf-8") return json.loads(body).get("versions", []) except Exception as e: 17121.m.kuaisou.com logger.warning(f"Fetch history failed: {e}") return [] # ==================== 配置客户端主体 ==================== class ConfigClient: 17122.m.kuaisou.com """分布式配置中心客户端""" def __init__( self, server_url: str = DEFAULT_SERVER_URL, namespace: str = DEFAULT_NAMESPACE, cache_dir: str = DEFAULT_CACHE_DIR, poll_timeout: int = DEFAULT_POLL_TIMEOUT, max_retries: int = DEFAULT_MAX_RETRIES, retry_backoff: float = DEFAULT_RETRY_BACKOFF, ): self.server_url = server_url self.default_namespace = namespace self.poll_timeout = poll_timeout self.max_retries = max_retries self.retry_backoff = retry_backoff self.http = ConfigHttpClient(server_url=server_url) self.cache = LocalCacheManager(cache_dir=cache_dir) self._configs: Dict[str, ConfigItem] = {} self._listeners: Dict[str, List[ListenerEntry]] = {} self._config_lock = threading.RLock() self._listener_lock = threading.Lock() self._running = False self._poll_thread: Optional[threading.Thread] = None self._callback_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="ConfigCallback") self.stats = { "fetches": 0, "cache_hits": 0, "updates": 0, "poll_cycles": 0, "poll_errors": 0, "callbacks": 0, } self._stats_lock = threading.Lock() def _inc_stat(self, key: str, delta: int = 1): 17129.m.kuaisou.com with self._stats_lock: self.stats[key] = self.stats.get(key, 0) + delta # ---------- 配置读取 ---------- def get_config( self, data_id: str, group: str = DEFAULT_GROUP, namespace: Optional[str] = None, fmt: str = CONFIG_FORMAT_JSON, ) -> Optional[Dict[str, Any]]: 17135.m.kuaisou.com ns = namespace or self.default_namespace cache_key = f"{ns}/{group}/{data_id}" # 内存缓存 with self._config_lock: 17137.m.kuaisou.com item = self._configs.get(cache_key) if item: self._inc_stat("cache_hits") return item.parsed # 远程拉取（带重试） item = self._fetch_with_retry(ns, group, data_id, fmt) if item: 17148.m.kuaisou.com with self._config_lock: 17147.m.kuaisou.com self._configs[cache_key] = item self.cache.save(item) return item.parsed # 降级到本地缓存 cached = self.cache.load(ns, group, data_id) if cached: 17152.m.kuaisou.com logger.info(f"Fallback to local cache: {cache_key}") with self._config_lock: 17153.m.kuaisou.com self._configs[cache_key] = cached return cached.parsed return None def get_raw_config( self, data_id: str, group: str = DEFAULT_GROUP, namespace: Optional[str] = None, ) -> Optional[str]: 17154.m.kuaisou.com ns = namespace or self.default_namespace cache_key = f"{ns}/{group}/{data_id}" with self._config_lock: 17157.m.kuaisou.com item = self._configs.get(cache_key) if item: 17162.m.kuaisou.com return item.content item = self._fetch_with_retry(ns, group, data_id) if item: 17168.m.kuaisou.com with self._config_lock: 17171.m.kuaisou.com self._configs[cache_key] = item return item.content cached = self.cache.load(ns, group, data_id) return cached.content if cached else None def _fetch_with_retry( self, namespace: str, group: str, data_id: str, fmt: str = CONFIG_FORMAT_JSON ) -> Optional[ConfigItem]: 17173.m.kuaisou.com for attempt in range(1, self.max_retries + 1): 17174.m.kuaisou.com self._inc_stat("fetches") result = self.http.fetch_config(namespace, group, data_id) if result and result.get("code") == 0: 17175.m.kuaisou.com data = result.get("data", {}) return ConfigItem( namespace=namespace, group=group, data_id=data_id, content=data.get("content", ""), md5=data.get("md5", ""), version=data.get("version", 0), fmt=fmt, last_modified=data.get("lastModified", time.time()), ) if attempt bool: 17197.m.kuaisou.com with self._listener_lock: 17207.m.kuaisou.com listening = [] for key, entries in self._listeners.items(): 22016.m.kuaisou.com active_entries = [e for e in entries if e.enabled] if not active_entries: 22021.m.kuaisou.com continue first = active_entries[0] listening.append({ "namespace": first.namespace, "group": first.group, "dataId": first.data_id, "md5": first.last_md5, }) if not listening: 26025.m.kuaisou.com return False changed_list = self.http.long_poll(listening, timeout=self.poll_timeout) if not changed_list: 26026.m.kuaisou.com return False updated = False for change in changed_list: 26028.m.kuaisou.com ns = change.get("namespace", self.default_namespace) group = change.get("group", DEFAULT_GROUP) data_id = change.get("dataId", "") new_md5 = change.get("md5", "") item = self._fetch_with_retry(ns, group, data_id) if not item: 26029.m.kuaisou.com continue cache_key = f"{ns}/{group}/{data_id}" with self._config_lock: 26030.m.kuaisou.com old_item = self._configs.get(cache_key) self._configs[cache_key] = item self.cache.save(item) self._inc_stat("updates") # 触发回调 with self._listener_lock: 26031.m.kuaisou.com entries = self._listeners.get(cache_key, []) for entry in entries: 26032.m.kuaisou.com if entry.enabled and entry.last_md5 != new_md5: 26033.m.kuaisou.com entry.last_md5 = new_md5 self._callback_executor.submit(self._safe_callback, entry, item) updated = True logger.info(f"Config updated: {cache_key} v{item.version} md5={new_md5[:8]}") return updated def _safe_callback(self, entry: ListenerEntry, item: ConfigItem): try: entry.callback(item) self._inc_stat("callbacks") except Exception as e: 26034.m.kuaisou.com logger.error(f"Callback error for {item.cache_key()}: {e}\n{traceback.format_exc()}") # ---------- 版本与回滚 ---------- def get_history( self, data_id: str, group: str = DEFAULT_GROUP, namespace: Optional[str] = None, limit: int = 10, ) -> List[Dict[str, Any]]: 26035.m.kuaisou.com ns = namespace or self.default_namespace return self.http.fetch_history(ns, group, data_id, limit=limit) def rollback_to_version( self, data_id: str, version: int, group: str = DEFAULT_GROUP, namespace: Optional[str] = None, ) -> bool: 26036.m.kuaisou.com ns = namespace or self.default_namespace history = self.http.fetch_history(ns, group, data_id, limit=50) target = None for h in history: 26037.m.kuaisou.com if h.get("version") == version: 26038.m.kuaisou.com target = h break if not target: 26039.m.kuaisou.com logger.warning(f"Version {version} not found for {data_id}") return False item = ConfigItem( namespace=ns, group=group, data_id=data_id, content=target.get("content", ""), md5=target.get("md5", ""), version=version, fmt=target.get("fmt", CONFIG_FORMAT_JSON), ) cache_key = f"{ns}/{group}/{data_id}" with self._config_lock: 26040.m.kuaisou.com self._configs[cache_key] = item self.cache.save(item) logger.info(f"Rolled back {cache_key} to v{version}") return True # ---------- 监控 ---------- def get_stats(self) -> Dict[str, Any]: 26042.m.kuaisou.com with self._stats_lock: 26041.m.kuaisou.com stats = dict(self.stats) with self._config_lock: 26043.m.kuaisou.com stats["cached_configs"] = len(self._configs) with self._listener_lock: 26045.m.kuaisou.com stats["active_listeners"] = sum(len(v) for v in self._listeners.values()) return stats def refresh_all(self): 26046.m.kuaisou.com """手动刷新所有已缓存的配置""" with self._config_lock: 26047.m.kuaisou.com keys = list(self._configs.keys()) for cache_key in keys: 26048.m.kuaisou.com parts = cache_key.split("/") if len(parts) == 3: 26049.m.kuaisou.com ns, group, data_id = parts item = self._fetch_with_retry(ns, group, data_id) if item: with self._config_lock: self._configs[cache_key] = item self.cache.save(item) logger.info(f"Refreshed {len(keys)} configs") ``` ### 架构要点说明 1. **长轮询变更监听**：客户端将当前持有的所有监听配置的 MD5 发送到服务端，服务端挂起请求直到有变更或超时。相比短轮询大幅降低无效请求，相比 WebSocket 兼容性更好。`long_poll` 的超时设置为 `poll_timeout + read_timeout`，防止网络层提前断开。 1. **三级读取降级**：`get_config` 依次尝试内存缓存 → 远程拉取（带指数退避重试）→ 本地文件缓存。即使服务端完全不可用，应用仍可启动并使用上次缓存的配置，保障可用性。 1. **原子文件缓存**：写入时先写 `.tmp` 文件再 `rename`，避免进程崩溃导致缓存文件损坏。缓存以 JSON 存储元数据+原始内容，加载时可重新解析，兼容格式升级。 1. **异步回调隔离**：监听器回调通过独立线程池执行，避免用户回调阻塞轮询主循环。每个回调包裹 `_safe_callback` 捕获异常，单个监听器出错不影响其他。 1. **命名空间隔离**：所有操作均携带 `namespace/group/dataId` 三元组，天然支持多租户、多环境隔离。缓存键和文件名均包含完整路径，避免冲突。 1. **版本管理与回滚**：通过历史记录 API 获取版本列表，`rollback_to_version` 将指定版本内容加载到内存和缓存，配合监听器自动触发热更新回调，实现无重启回滚。
2026-10-20
优秀的个人博客,低调大师

Python 爬虫与反爬技术实战：从基础采集到 TLS 指纹对抗

反爬与爬虫的对抗在 2026 年已经进入深水区。单纯改个 User-Agent 或加个 `time.sleep` 早已不够用，现代风控系统会在 TLS 握手阶段就识别出 `requests` 库的非浏览器指纹，哪怕你用的是干净的住宅 IP，照样秒封。本文从可运行的基础爬虫代码讲起，逐层拆解频率控制、IP 轮换、TLS 指纹伪装与浏览器自动化对抗，所有代码均可直接运行或稍作配置后使用。 --- ## 基础爬虫：带重试与随机间隔的多页采集 先建立一个“能跑且不容易把自己跑死”的基线。以下代码使用 `requests.Session` 复用连接，配合 `urllib3` 的重试机制处理瞬时网络波动，并引入随机延时打破固定频率特征。 ``` import csv import random import time from urllib.parse import urljoin import requests from bs4 import BeautifulSoup from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE_URL = "https://www.cza9.com/" OUTPUT_FILE = "products.csv" session = requests.Session() session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/126.0.0.0 Safari/537.36" ), "Accept-Language": "zh-CN,zh;q=0.9", }) retry = Retry( total=2, connect=2, read=2, status=2, backoff_factor=0.8, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"], ) adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) def fetch_page(page_number): url = BASE_URL.format(page_number) # 随机间隔 0.8~2.5 秒，避免固定频率被识别 time.sleep(random.uniform(0.8, 2.5)) resp = session.get(url, timeout=(5, 15)) resp.raise_for_status() return resp.text, url def parse_products(html, page_url): soup = BeautifulSoup(html, "html.parser") items = [] for card in soup.select("article.product_pod"): title_tag = card.select_one("h3 a") price_tag = card.select_one(".price_color") if not title_tag or not price_tag: continue items.append({ "title": title_tag.get("title", "").strip(), "price": price_tag.get_text(strip=True), "url": urljoin(page_url, title_tag.get("href", "")), }) return items def main(max_pages=5): all_items = [] for page in range(1, max_pages + 1): try: html, url = fetch_page(page) items = parse_products(html, url) all_items.extend(items) print(f"第 {page} 页：获取 {len(items)} 条") except Exception as e: print(f"第 {page} 页失败：{e}") with open(OUTPUT_FILE, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["title", "price", "url"]) writer.writeheader() writer.writerows(all_items) print(f"共写入 {len(all_items)} 条到 {OUTPUT_FILE}") if __name__ == "__main__": main() ``` **关键点**：`Retry` 的重试次数不宜超过 3 次。如果连续收到 `403` 或 `429`，说明已经触发风控，继续重试只会加深封禁标记，应当停下来检查策略。 --- ## 反爬第一层：IP 封锁与代理轮换 单一 IP 高频访问是触发封禁的最常见原因。反爬系统不仅看当前秒级的请求数，还会维护 IP 信誉画像——这个 IP 是否只访问详情页、是否从未加载过静态资源、是否触发过验证码。数据中心 IP（如阿里云、腾讯云、AWS 的 IP 段）在 ASN 层面就会被标记为“半信任”，住宅代理的存活率明显更高。 以下是一个带有效性检测的代理轮换示例： ``` import requests from itertools import cycle proxy_pool = [ "http://user:pass@res-ip-1.example.com:8080", "http://user:pass@res-ip-2.example.com:8080", ] proxy_cycle = cycle(proxy_pool) def request_with_proxy(url, max_attempts=3): for attempt in range(max_attempts): proxy = next(proxy_cycle) try: resp = requests.get( url, proxies={"http": proxy, "https": proxy}, headers={"User-Agent": "Mozilla/5.0 ..."}, timeout=15, ) if resp.status_code == 403: # 403 立即换代理，不要用原 IP 重试 continue resp.raise_for_status() return resp.text except requests.RequestException: continue raise RuntimeError(f"经过 {max_attempts} 次代理尝试仍失败") ``` 对于需要保持登录态的场景，不能每个请求都换 IP。服务端 session 往往绑定了 IP，轮换太快会导致 `401`。此时应使用 **sticky session**（同一 IP 保持 30 分钟以上），而非按请求轮换。 --- ## 反爬第二层：TLS 指纹识别与 `curl_cffi` 对抗 这是 2026 年最容易被忽视的拦截点。即使你用了住宅代理、补全了所有 HTTP 头，`requests` 发出的 TLS Client Hello 包中，加密套件顺序、扩展列表、椭圆曲线参数与真实 Chrome 存在明显差异。风控系统在 TLS 握手阶段计算 JA3/JA4 指纹，一旦与声明的 User-Agent 不匹配，直接拒绝。 **实测验证**：你可以访问 `https://tls.peet.ws/api/all` 查看 `requests` 发出的真实 JA3 指纹，再与 Chrome 对比。两者几乎不可能一致。 解决方案是使用 `curl_cffi`，它在 C 层面模拟浏览器的 TLS 握手： ``` # pip install curl_cffi from curl_cffi import requests response = requests.get( "https://tls.peet.ws/api/all", impersonate="chrome120", # 模拟 Chrome 120 的 TLS 指纹与 HTTP/2 行为 proxies={ "http": "http://user:pass@proxy.example.com:8080", "https": "http://user:pass@proxy.example.com:8080", }, timeout=15, ) data = response.json() print("JA3:", data.get("tls", {}).get("ja3_hash")) print("User-Agent 一致性:", data.get("http2", {}).get("akamai_fingerprint")) ``` `impersonate` 参数支持 `chrome120`、`chrome116`、`safari15_5`、`edge101`、`firefox120` 等。选择与你 User-Agent 版本匹配的参数，否则 JA3 与 UA 不一致反而更容易被识别。 --- ## 反爬第三层：JavaScript 渲染与浏览器自动化 当页面数据由 JavaScript 动态加载，或者目标站部署了 Cloudflare 的 JS Challenge（5 秒盾）时，纯 HTTP 请求无法获取完整内容。此时需要浏览器自动化工具。 Playwright 是当前推荐的选择，相比 Selenium 对现代浏览器的支持更完整。以下代码启动 Chromium 并提取动态渲染后的内容： ``` # pip install playwright && playwright install chromium from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() # 设置合理的视口和 UA，减少无头特征 page.set_viewport_size({"width": 1920, "height": 1080}) page.goto("https://www.cza9.com", wait_until="networkidle") # 等待动态内容出现 page.wait_for_selector(".results", timeout=15000) content = page.content() print(content[:500]) browser.close() ``` 但仅靠 Playwright 原生 API 仍然会被检测到 `navigator.webdriver = true` 等特征。可以使用 `playwright-stealth` 注入脚本来掩盖这些痕迹： ``` # pip install playwright-stealth from playwright.sync_api import sync_playwright from playwright_stealth import stealth_sync with sync_playwright() as p: browser = p.chromium.launch(headless=False) # 有头模式存活率更高 page = browser.new_page() stealth_sync(page) # 一行调用，隐藏 webdriver 等自动化特征 page.goto("https://wap.cza9.com") browser.close() ``` 需要注意：stealth 插件主要应对基础的 `navigator.webdriver` 检测和部分 Canvas 指纹混淆。面对 Cloudflare 企业级防护或 DataDome 时，它通常只能争取到“被放行一次”的机会，难以维持长期稳定采集。 --- ## 反爬第四层：行为分析与随机化策略 即使指纹和 IP 都过关，行为模式仍然会暴露脚本身份。固定间隔的 `sleep(1)` 连续执行数小时，在风控系统看来比随机间隔更可疑——真人不会精确到毫秒级重复操作。 几个容易被忽略的行为细节： **滚动与鼠标移动**：Playwright 的 `mouse.wheel` 和 `mouse.move` 可以模拟真实滚动和悬停，但需要加入随机增量： ``` import random # 随机滚动，模拟阅读 for _ in range(random.randint(2, 5)): page.mouse.wheel(0, random.randint(300, 800)) page.wait_for_timeout(random.randint(500, 2000)) ``` **页面停留时间**：在关键操作（如翻页、点击详情）之间加入非均匀分布的等待。简单的 `random.uniform` 已经比固定值好，但更自然的方式是使用对数正态分布，让短等待出现频率更高、偶尔出现较长停顿。 **Cookie 持久化**：不要每次启动都从零开始。使用 `storage_state` 保存会话，在后续运行中恢复，可以积累“信任分”： ``` context = browser.new_context(storage_state="session.json") # ... 执行采集 ... context.storage_state(path="session.json") # 保存 ``` --- ## 合规与边界 所有技术手段都应以**获得授权或目标站明确允许**为前提。实操前至少确认三件事： 1. **查看 `robots.txt**`：虽然它不具法律强制力，但违反明确的禁止声明可能构成违约或侵权。 1. **控制请求量级**：哪怕技术能绕过限制，也不应以远超正常用户的速度施压。`429` 是服务器在说“你太快了”，尊重这个信号。 1. **不触碰个人隐私数据**：手机号、地址、身份证号等字段的采集在绝大多数司法辖区都有额外合规要求。 反爬技术是一把双刃剑，用于公开数据的合理采集与用于绕过保护机制盗取敏感数据，边界在于授权与目的。代码能跑通不等于应该跑。
2026-10-20

资源下载

更多资源
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

时间: 2026-08-21 大小: 1MB 下载: 10次
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2026-09-24 大小: 189MB 下载: 9次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-08-21 大小: 1.42GB 下载: 6次
Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

时间: 2026-08-21 大小: 9.84MB 下载: 40次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册