系统稳定性堪称软件的“生命线”。无论功能多么丰富，如果频繁出现宕机、报错或卡顿，用户体验将急剧下降。那么如何实现高稳定性？核心在于掌握关键指标与成熟的工程模式。深入理解这些内容，就能从容应对绝大多数场景。

2.1 稳定性核心指标

评估系统稳定性的最直接指标是SLA（服务等级协议）。它通过“多少个9”来量化可用性，例如99.9%、99.99%——每增加一个9，允许的停机时间便大幅缩减。下图中清晰展示了SLA与允许停机时间的对应关系：

SLA与允许停机时间对照表：

了解了这些基础知识，我们再来探讨常见的容错设计模式。

2.2 容错设计模式

稳定性并非空谈，而是由一系列扎实的工程模式支撑起来的。以下是实际项目中最常用且最具效用的几种模式。

超时与重试机制

设想一下，系统向外部发送一个请求，但对方迟迟没有响应。如果缺乏超时控制，该请求将永久挂起，逐渐耗尽线程池和连接池，最终导致系统崩溃。因此，超时机制是必不可少的兜底手段。

更完善的方案是在超时基础上加入重试功能。然而，重试并非盲目循环等待，通常采用指数退避策略——首次失败等待1秒，第二次等待2秒，第三次等待4秒，依此类推。这样既能为目标服务留出恢复时间，又能避免“重试风暴”的发生。

class ResilientClient:
    """支持超时与重试的容错客户端"""
    def __init__(self, timeout=5, max_retries=3, backoff_factor=2):
        self.timeout = timeout
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor

    async def call_with_retry(self, func, *args, **kwargs):
        """带超时控制和重试机制的远程调用"""
        last_exception = None
        for attempt in range(self.max_retries):
            try:
                # 设置超时
                result = await asyncio.wait_for(func(*args, **kwargs),
                                                timeout=self.timeout)
                return result
            except asyncio.TimeoutError:
                last_exception = TimeoutError(f"Timeout after {self.timeout}s")
                logging.warning(f"Attempt {attempt   1} timed out")
            except Exception as e:
                last_exception = e
                logging.warning(f"Attempt {attempt   1} failed: {e}")

            # 最后一次不等待
            if attempt < self.max_retries - 1:
                wait_time = self.backoff_factor ** attempt
                logging.info(f"Retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)

        raise last_exception

仅有超时和重试仍然不够，另一个关键组件是断路器模式。它类似于电路保险丝——当下游服务频繁出错时，断路器会果断断开，阻止请求继续流入。等待一段时间（例如60秒）后，再尝试放行少量请求进行“探测”。如果恢复成功，断路器闭合；如果继续失败，则再次断开。这一机制能够有效防止级联故障，避免一个小问题演变为全网雪崩。

class CircuitBreaker:
    """断路器：防止级联故障的容错组件"""
    def __init__(self, failure_threshold=5, timeout=60, half_open_requests=3):
        self.failure_threshold = failure_threshold       # 失败阈值
        self.timeout = timeout                           # 恢复超时
        self.half_open_requests = half_open_requests     # 半开状态请求数
        self.failures = 0
        self.last_failure_time = None
        self.state = "CLOSED"                            # CLOSED, OPEN, HALF_OPEN
        self.half_open_successes = 0

    async def call(self, func, *args, **kwargs):
        """带断路器保护的远程调用"""
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
                self.half_open_successes = 0
                logging.info("Circuit breaker moved to HALF_OPEN")
            else:
                raise CircuitBreakerOpenError("Circuit breaker is OPEN")

        try:
            result = await func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.half_open_successes  = 1
                if self.half_open_successes >= self.half_open_requests:
                    self.state = "CLOSED"
                    self.failures = 0
                    logging.info("Circuit breaker closed (recovered)")
            return result
        except Exception as e:
            self.failures  = 1
            self.last_failure_time = time.time()
            if self.state == "CLOSED" and self.failures >= self.failure_threshold:
                self.state = "OPEN"
                logging.error(f"Circuit breaker opened after {self.failures} failures")
            raise e

限流与降级策略

系统能够承载的流量存在上限，超出上限时必须学会“拒绝”。限流常用的策略是令牌桶算法：以固定速率向桶中放入令牌，每个请求到来时取走一个——有令牌则放行，无令牌则直接拒绝。该方法既能平滑突发流量，又能有效控制整体请求速率。

然而限流只是第一步。当流量真正达到极限时，还需要考虑服务降级——即舍弃部分非核心功能，确保核心逻辑可用。例如，推荐系统负载过高时，可以直接返回一组固定的热门推荐，而不是强行计算个性化推荐。这种“有总比没有好”的兜底策略，在实际生产中非常实用。

class RateLimiter:
    """基于令牌桶算法的限流器"""
    def __init__(self, rate=100, capacity=200):
        self.rate = rate                    # 令牌生成速率（每秒）
        self.capacity = capacity            # 桶容量
        self.tokens = capacity              # 当前令牌数
        self.last_refill = time.time()
        self._lock = threading.Lock()

    def acquire(self, tokens=1):
        """获取令牌，成功返回True"""
        with self._lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    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

# 降级策略
class DegradationStrategy:
    """服务降级策略实现"""
    def __init__(self, fallback_func):
        self.fallback_func = fallback_func
        self.degraded = False
        self.degraded_until = None

    @property
    def is_degraded(self):
        if self.degraded_until and time.time() > self.degraded_until:
            self.degraded = False
            self.degraded_until = None
        return self.degraded

    def degrade(self, duration=60):
        """进入降级模式"""
        self.degraded = True
        self.degraded_until = time.time()   duration
        logging.warning(f"Service degraded for {duration}s")

    def execute(self, primary_func, *args, **kwargs):
        """执行主逻辑或降级逻辑"""
        if self.is_degraded:
            return self.fallback_func(*args, **kwargs)
        try:
            return primary_func(*args, **kwargs)
        except Exception as e:
            logging.error(f"Primary function failed: {e}")
            self.degrade()
            return self.fallback_func(*args, **kwargs)

# 使用示例
def get_recommendations_fallback(user_id):
    """降级方案：返回默认推荐"""
    return DEFAULT_RECOMMENDATIONS

@rate_limit(rate=50, capacity=100)
def get_recommendations(user_id):
    recommendations = recommendation_service.get(user_id)
    return recommendations

# 包装降级
strategy = DegradationStrategy(get_recommendations_fallback)
result = strategy.execute(get_recommendations, user_id)

幂等性设计

还有一个经常被忽略但至关重要的设计——幂等性。简而言之，同一请求无论执行多少次，其结果都应保持一致。例如支付接口，用户点击两次“付款”，绝不能扣除两次款项。

实现思路并不复杂：为每个请求分配一个唯一ID，服务端在处理前先检查该ID是否已被处理过。若已处理，则直接返回上次的结果；若未处理，则正常执行并缓存结果。这一机制能有效避免重试带来的副作用，使系统更加健壮。

class IdempotentHandler:
    """幂等性处理器，防止重复处理"""
    def __init__(self, redis_client, ttl=86400):
        self.redis = redis_client
        self.ttl = ttl

    def idempotent(self, key_prefix):
        """幂等性装饰器"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 生成唯一请求ID
                request_id = kwargs.get('request_id') or str(uuid.uuid4())
                idempotent_key = f"{key_prefix}:{request_id}"

                # 检查是否已处理
                result = self.redis.get(idempotent_key)
                if result is not None:
                    logging.info(f"Duplicate request: {request_id}")
                    return json.loads(result)

                # 执行处理
                result = func(*args, **kwargs)

                # 缓存结果
                self.redis.setex(idempotent_key,
                                 self.ttl,
                                 json.dumps(result))
                return result
            return wrapper
        return decorator

# 使用示例
handler = IdempotentHandler(redis_client)

@handler.idempotent("payment")
def process_payment(order_id, amount, request_id=None):
    """处理支付 - 同一request_id只会执行一次"""
    # 扣款逻辑
    return {"status": "success", "transaction_id": str(uuid.uuid4())}

2.3 故障演练与混沌工程

最后探讨一个更高级的话题：如何验证你的容错机制是否真正可靠？答案就是——主动“制造故障”，在测试环境中提前演练。这正是混沌工程的核心思想。

例如，你可以故意向某个服务注入延迟（使接口响应变慢几百毫秒），或随机返回错误码，然后观察整个系统的反应。如果断路器正常断开、降级机制及时生效，则说明设计达标。反之，若系统直接崩溃，则表明存在需要改进的薄弱环节。

这种“先破后立”的策略，将稳定性问题暴露在可控的测试环境中，远比在生产环境中“赌运气”更加可靠。混沌工程并非为了破坏，而是通过实验手段验证系统的韧性。

# 混沌实验配置
class ChaosExperiment:
    """混沌实验执行器，用于注入故障并验证系统韧性"""
    def __init__(self):
        self.experiments = []

    def add_latency_injection(self, service, latency_ms=100, probability=0.1):
        """注入延迟"""
        self.experiments.append({
            "type": "latency",
            "service": service,
            "latency_ms": latency_ms,
            "probability": probability
        })

    def add_error_injection(self, service, error_code=500, probability=0.05):
        """注入错误"""
        self.experiments.append({
            "type": "error",
            "service": service,
            "error_code": error_code,
            "probability": probability
        })

    def execute(self):
        """执行混沌实验"""
        for exp in self.experiments:
            asyncio.create_task(self._run_experiment(exp))

    async def _run_experiment(self, exp):
        while True:
            if random.random() < exp["probability"]:
                await self._inject_fault(exp)
            await asyncio.sleep(1)

    async def _inject_fault(self, exp):
        if exp["type"] == "latency":
            # 注入网络延迟
            await asyncio.sleep(exp["latency_ms"] / 1000)
        elif exp["type"] == "error":
            # 注入错误响应
            raise HTTPException(status_code=exp["error_code"])

# 在测试环境运行混沌实验
if os.getenv("CHAOS_ENABLED") == "true":
    chaos = ChaosExperiment()
    chaos.add_latency_injection("database", latency_ms=200, probability=0.05)
    chaos.add_error_injection("payment_service", error_code=503, probability=0.02)
    chaos.execute()