3.2.2 限流 / 熔断 / 降级 / 隔离(Sentinel / Hystrix / Resilience4j)

import time
import threading

class TokenBucket:
    """令牌桶限流器 — 最常用,允许突发"""
    def __init__(self, rate: float, capacity: int):
        self.rate = rate           # 每秒生成令牌数
        self.capacity = capacity   # 桶容量
        self.tokens = capacity     # 当前令牌数
        self.last_refill = time.time()
        self.lock = threading.Lock()
    
    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        # 按时间补充令牌
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now
    
    def try_acquire(self, n: int = 1) -> bool:
        with self.lock:
            self._refill()
            if self.tokens >= n:
                self.tokens -= n
                return True
            return False

# 测试:100 QPS,桶容量 200
bucket = TokenBucket(rate=100, capacity=200)
allowed = sum(bucket.try_acquire() for _ in range(500))
print(f"允许 {allowed} / 500")  # 大约 300(初始 200 + 100 个时间生成)

import time
import threading
from collections import deque

class LeakyBucket:
    """漏桶限流器 — 强制固定速率,绝对平滑"""
    def __init__(self, rate: float, capacity: int):
        self.rate = rate           # 每秒漏出数
        self.capacity = capacity   # 桶容量
        self.queue = deque()       # 请求队列
        self.last_leak = time.time()
        self.lock = threading.Lock()
    
    def try_acquire(self) -> bool:
        with self.lock:
            now = time.time()
            elapsed = now - self.last_leak
            leaked = elapsed * self.rate
            # 漏出已处理的请求
            while self.queue and leaked > 0:
                leaked -= 1
                self.queue.popleft()
            self.last_leak = now
            
            if len(self.queue) < self.capacity:
                self.queue.append(now)
                return True
            return False

# 测试:100 QPS,桶容量 100
bucket = LeakyBucket(rate=100, capacity=100)
allowed = sum(bucket.try_acquire() for _ in range(500))
print(f"允许 {allowed} / 500")  # 大约 100(初始) + 时间补充

import time

class FixedWindow:
    """固定窗口限流器 — 简单但有突刺问题"""
    def __init__(self, limit: int, window_sec: int = 1):
        self.limit = limit
        self.window_sec = window_sec
        self.count = 0
        self.window_start = time.time()
    
    def try_acquire(self) -> bool:
        now = time.time()
        if now - self.window_start >= self.window_sec:
            self.count = 0
            self.window_start = now
        if self.count < self.limit:
            self.count += 1
            return True
        return False

class SlidingWindow:
    """滑动窗口限流器 — 精度高,内存稍多"""
    def __init__(self, limit: int, window_sec: int = 1, buckets: int = 10):
        self.limit = limit
        self.bucket_sec = window_sec / buckets
        self.buckets = [0] * buckets
        self.bucket_start = [0.0] * buckets
    
    def try_acquire(self) -> bool:
        import time
        now = time.time()
        idx = int(now / self.bucket_sec) % len(self.buckets)
        
        # 过期清零
        if now - self.bucket_start[idx] >= self.bucket_sec * len(self.buckets):
            self.buckets[idx] = 0
            self.bucket_start[idx] = now
        
        if sum(self.buckets) < self.limit:
            self.buckets[idx] += 1
            return True
        return False

import redis
import time

r = redis.Redis(host='localhost', port=6379)
SCRIPT = """
local key = KEYS[1]
local window = tonumber(ARGV[1])
local limit = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local cutoff = now - window
redis.call('ZREMRANGEBYSCORE', key, '-inf', cutoff)
local count = redis.call('ZCARD', key)
if count < limit then
    redis.call('ZADD', key, now, now .. ':' .. math.random())
    redis.call('PEXPIRE', key, window)
    return 1
else
    return 0
end
"""

def rate_limit(user_id: str, limit: int = 100, window_ms: int = 1000) -> bool:
    """分布式限流:Redis Lua 原子操作"""
    key = f"rate:{user_id}"
    now = int(time.time() * 1000)
    result = r.eval(SCRIPT, 1, key, window_ms, limit, now)
    return result == 1

# 调用
if rate_limit("user_123", limit=100):
    handle_request()
else:
    return "Too Many Requests", 429

import time
import threading
from enum import Enum
from collections import deque
from functools import wraps

class State(Enum):
    CLOSED = "CLOSED"
    OPEN = "OPEN"
    HALF_OPEN = "HALF_OPEN"

class CircuitBreaker:
    """熔断器 — Hystrix 风格"""
    def __init__(
        self,
        failure_threshold: int = 5,      # 失败次数阈值
        recovery_timeout: float = 10.0,  # 熔断恢复时间(秒)
        success_threshold: int = 3,      # Half-Open 成功几次后关闭
        time_window: float = 10.0,       # 统计窗口(秒)
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.time_window = time_window
        
        self.state = State.CLOSED
        self.failures = deque()  # (timestamp, exception)
        self.successes_in_half_open = 0
        self.opened_at = 0.0
        self.lock = threading.Lock()
    
    def call(self, func, *args, **kwargs):
        with self.lock:
            # 1. 检查状态
            if self.state == State.OPEN:
                if time.time() - self.opened_at >= self.recovery_timeout:
                    self.state = State.HALF_OPEN
                    self.successes_in_half_open = 0
                else:
                    raise Exception("Circuit breaker is OPEN")
        
        # 2. 执行调用
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure(e)
            raise
    
    def _on_success(self):
        with self.lock:
            if self.state == State.HALF_OPEN:
                self.successes_in_half_open += 1
                if self.successes_in_half_open >= self.success_threshold:
                    self.state = State.CLOSED
                    self.failures.clear()
            elif self.state == State.CLOSED:
                # 清理过期失败记录
                cutoff = time.time() - self.time_window
                while self.failures and self.failures[0][0] < cutoff:
                    self.failures.popleft()
    
    def _on_failure(self, exception):
        with self.lock:
            now = time.time()
            if self.state == State.HALF_OPEN:
                self.state = State.OPEN
                self.opened_at = now
            elif self.state == State.CLOSED:
                cutoff = now - self.time_window
                while self.failures and self.failures[0][0] < cutoff:
                    self.failures.popleft()
                self.failures.append((now, exception))
                if len(self.failures) >= self.failure_threshold:
                    self.state = State.OPEN
                    self.opened_at = now

# 使用示例
@CircuitBreaker(failure_threshold=5, recovery_timeout=10)
def call_payment_service(order_id):
    # 远程调用
    return requests.post(...)

# 降级函数
def fallback(order_id):
    return {"order_id": order_id, "status": "pending", "msg": "支付排队中"}

class SentinelCircuitBreaker:
    """Sentinel 风格熔断器:基于错误率 + 慢调用率"""
    def __init__(
        self,
        error_ratio_threshold: float = 0.5,    # 错误率阈值 50%
        slow_ratio_threshold: float = 0.3,     # 慢调用率阈值 30%
        slow_call_timeout_ms: int = 1000,      # 慢调用判定
        min_request_amount: int = 10,          # 最小请求数(避免抖动)
        recovery_timeout_ms: int = 30000,      # 熔断恢复时间
        stat_window_ms: int = 10000,           # 统计窗口
    ):
        self.error_ratio = error_ratio_threshold
        self.slow_ratio = slow_ratio_threshold
        self.slow_timeout = slow_call_timeout_ms / 1000
        self.min_amount = min_request_amount
        self.recovery_timeout = recovery_timeout_ms / 1000
        self.stat_window = stat_window_ms / 1000
        
        self.total = 0
        self.errors = 0
        self.slow = 0
        self.window_start = time.time()
        self.state = State.CLOSED
        self.opened_at = 0.0
    
    def entry(self) -> bool:
        if self.state == State.OPEN:
            if time.time() - self.opened_at >= self.recovery_timeout:
                self.state = State.HALF_OPEN
                return True
            return False
        return True
    
    def exit(self, success: bool, duration: float):
        self.total += 1
        if not success:
            self.errors += 1
        if duration > self.slow_timeout:
            self.slow += 1
        
        # 窗口结束,判断是否熔断
        if time.time() - self.window_start >= self.stat_window:
            self._check_and_break()
            self._reset_window()
    
    def _check_and_break(self):
        if self.total < self.min_amount:
            return
        error_ratio = self.errors / self.total
        slow_ratio = self.slow / self.total
        if error_ratio >= self.error_ratio or slow_ratio >= self.slow_ratio:
            self.state = State.OPEN
            self.opened_at = time.time()
    
    def _reset_window(self):
        self.total = self.errors = self.slow = 0
        self.window_start = time.time()

class DegradationManager:
    """降级管理器:配置中心 + 自动恢复"""
    def __init__(self, config_client):
        self.config = config_client
        self.switches = {}
        self.last_check = {}
    
    def is_degraded(self, service: str) -> bool:
        """查询是否降级"""
        return self.switches.get(service, {}).get('enabled', False)
    
    def execute_with_fallback(self, service: str, primary_func, fallback_func, *args):
        """带降级的执行"""
        if self.is_degraded(service):
            return fallback_func(*args)
        try:
            return primary_func(*args)
        except Exception:
            # 触发降级
            self._trigger(service)
            return fallback_func(*args)
    
    def _trigger(self, service: str):
        """触发降级(自动恢复)"""
        conf = self.switches.get(service, {})
        if conf.get('auto_recovery', False):
            self.switches[service] = {'enabled': True, 'triggered_at': time.time()}
    
    def auto_recover(self):
        """定期检查自动恢复"""
        now = time.time()
        for service, conf in self.switches.items():
            if conf.get('enabled') and conf.get('triggered_at'):
                if now - conf['triggered_at'] > conf.get('recovery_interval', 60):
                    self.switches[service]['enabled'] = False

# 使用
deg = DegradationManager(config_client)

def call_recommend(user_id):
    return deg.execute_with_fallback(
        'recommendation-service',
        primary_func=remote_recommend,
        fallback_func=lambda uid: get_hot_items(),  # 返回热门商品
        user_id
    )

# 读降级:三级缓存
def get_product(product_id):
    # L1: 本地缓存
    p = local_cache.get(product_id)
    if p: return p
    # L2: Redis
    p = redis.get(f"product:{product_id}")
    if p: return p
    # L3: 数据库
    try:
        p = db.query(product_id)
        redis.setex(f"product:{product_id}", 300, p)
        return p
    except Exception:
        return {"id": product_id, "name": "已售罄", "price": 0}  # 降级

# 写降级:异步队列
def create_order(order):
    try:
        return db.create(order)
    except DBException:
        # 写失败,进 MQ 异步重试
        mq.send('order_create_retry', order)
        return {"status": "queued", "msg": "订单排队中"}

import threading
from concurrent.futures import ThreadPoolExecutor, TimeoutError

class BulkheadThreadPool:
    """线程池隔离 — Hystrix Command 风格"""
    def __init__(self, name: str, max_workers: int = 20, queue_size: int = 50):
        self.name = name
        self.executor = ThreadPoolExecutor(
            max_workers=max_workers,
            thread_name_prefix=f"bulkhead-{name}"
        )
        self.queue_size = queue_size
        self.semaphore = threading.Semaphore(max_workers + queue_size)
    
    def execute(self, func, *args, timeout: float = 5.0):
        if not self.semaphore.acquire(blocking=False):
            raise BulkheadFullException(f"{self.name} 线程池已满")
        try:
            future = self.executor.submit(func, *args)
            return future.result(timeout=timeout)
        except TimeoutError:
            raise TimeoutException(f"{self.name} 调用超时")
        finally:
            self.semaphore.release()

# 业务隔离
order_pool = BulkheadThreadPool("order", max_workers=20)
payment_pool = BulkheadThreadPool("payment", max_workers=20)

def call_payment(order_id):
    return payment_pool.execute(remote_pay, order_id, timeout=2.0)

import threading

class BulkheadSemaphore:
    """信号量隔离 — 轻量级,适合高频调用"""
    def __init__(self, name: str, max_concurrent: int = 100):
        self.name = name
        self.semaphore = threading.Semaphore(max_concurrent)
        self.active = 0
        self.lock = threading.Lock()
    
    def execute(self, func, *args, **kwargs):
        if not self.semaphore.acquire(blocking=False):
            raise BulkheadFullException(f"{self.name} 超过最大并发 {self.semaphore._value}")
        try:
            with self.lock:
                self.active += 1
            return func(*args, **kwargs)
        finally:
            with self.lock:
                self.active -= 1
            self.semaphore.release()

# 压测脚本(wrk 或 locust)
# locustfile.py
from locust import HttpUser, task

class StressUser(HttpUser):
    @task
    def order(self):
        self.client.post("/api/order", json={...})

# 启动:locust -f locustfile.py --users 1000 --spawn-rate 100
# 观察 P99 RT < 200ms 时的 QPS,即系统容量