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

原创

用户12583550

发布于 2026-10-21 18:23:45

190

在微服务架构中，服务间的级联故障（雪崩效应）是系统稳定性的致命威胁。下面从零实现一个生产级轻量级流量防护与高可用框架（类似 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.threshold

# ==================== 熔断器 ====================

@dataclass
class DegradeRule:
    """熔断规则"""
    strategy: DegradeStrategy
    threshold: float             # 阈值 (比例 0.0~1.0 或 数量)
    slow_ratio_threshold: float = 0.5 # 慢调用比例阈值
    max_slow_ms: float = 500.0   # 慢调用临界时间 (ms)
    min_request_amount: int = 5  # 最小请求数 (样本量不足不熔断)
    stat_interval_ms: int = 10000
    sleep_window_ms: int = 5000  # 熔断休眠时间

class CircuitBreaker: 17180.m.kuaisou.com 
    """熔断器状态机"""
    
    def __init__(self, rule: DegradeRule):
        self.rule = rule
        self.state = State.CLOSED
        self.metric = SlidingWindowMetric(interval_ms=rule.stat_interval_ms)
        self._lock = threading.Lock()
        self._next_retry_time = 0.0

    def on_request(self): 17178.m.kuaisou.com 
        """请求进入时的检查"""
        with self._lock: 17177.m.kuaisou.com    
            if self.state == State.CLOSED: 17183.m.kuaisou.com 
                return
            elif self.state == State.OPEN: 17184.m.kuaisou.com 
                if time.time() * 1000 >= 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.min_request_amount:
            return

        should_trip = False
        if self.rule.strategy == DegradeStrategy.ERROR_RATIO: 17193.m.kuaisou.com 
            if self.metric.get_error_ratio() >= 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

架构要点说明

滑动窗口指标统计 (Sliding Window)：摒弃了传统的固定窗口（存在临界点突刺问题）和单纯的漏桶算法。SlidingWindowMetric 将时间轴划分为多个等长的 MetricBucket（如 10 秒窗口分为 10 个 1 秒的桶）。通过淘汰过期桶、累加有效桶，实现了 O(1) 时间复杂度的高精度 QPS、错误率和慢调用比例统计，且内存占用极小。
令牌桶与平滑限流： TokenBucketLimiter 实现了经典的令牌桶算法。与固定窗口限流不同，令牌桶允许一定程度的突发流量（Burst）。当系统空闲时，令牌会累积到 max_burst 上限；当流量突增时，可以瞬间消耗累积的令牌，既保护了系统不被持续高压压垮，又充分利用了系统的空闲处理能力。
熔断器状态机 (Circuit Breaker)：严格实现了 Closed（闭合）、Open（断开）、Half-Open（半开）三态转换。
- Closed -> Open：当统计窗口内的错误率或慢调用比例超过阈值，且请求数达到最小样本量（防止偶发请求导致误熔断）时，触发熔断。
- Open -> Half-Open：经过 sleep_window_ms 休眠期后，自动转为半开状态，放行一个探测请求。
- Half-Open -> Closed/Open：若探测请求成功，则恢复闭合；若失败，则重新断开并重置休眠时间。
多维度降级策略：支持三种核心熔断策略：ERROR_RATIO（异常比例）、SLOW_REQUEST_RATIO（慢调用比例）和 ERROR_COUNT（异常数）。特别是慢调用比例熔断，能够有效防止下游服务响应变慢（但未报错）导致上游连接池耗尽的“隐性雪崩”问题。
Fallback 降级与异常隔离：通过 @protect 装饰器，业务代码只需关注核心逻辑。当触发限流（FlowException）时，执行 block_handler（如返回“系统繁忙”提示）；当触发业务异常或熔断（DegradeException）时，执行 fallback（如返回默认兜底数据或读取缓存）。实现了核心逻辑与容错逻辑的彻底解耦。
动态规则热更新： GuardManager 采用单例模式管理所有资源的守卫实例。规则（FlowRule / DegradeRule）与守卫实例解耦，支持在运行时通过 load_flow_rules 动态下发和更新规则（生产环境中可对接 Nacos/Apollo 等配置中心），无需重启服务即可调整限流阈值和熔断策略。