Python应用熔断降级怎么实现

wen python案例 2026-07-10 279

本文目录导读：

Python应用熔断降级怎么实现

使用 pybreaker 库（推荐）
使用 circuitbreaker 库
手动实现简单的熔断器
结合重试策略
配置化管理
最佳实践建议

在Python中实现熔断降级,最常用的方法是使用第三方库，以下是几种常见的实现方案：

使用 `pybreaker` 库（推荐）

pybreaker 是最流行的Python熔断器库之一。

安装

pip install pybreaker

基本使用

import pybreaker
import requests
from functools import wraps
# 创建熔断器
breaker = pybreaker.CircuitBreaker(
    fail_max=3,        # 失败3次后熔断
    reset_timeout=30,  # 30秒后尝试恢复
    state_storage=pybreaker.STATE_STORAGE_REGISTRY,
    exclude=[requests.ConnectionError]  # 排除某些异常不计入失败
)
# 方式1：使用装饰器
@breaker
def call_external_api():
    response = requests.get("https://api.example.com/data")
    return response.json()
# 方式2：使用上下文管理器
def get_data_with_breaker():
    try:
        with breaker:
            response = requests.get("https://api.example.com/data")
            return response.json()
    except pybreaker.CircuitBreakerError:
        # 熔断时的降级处理
        return {"message": "Service unavailable", "data": None}

高级配置

import pybreaker
import logging
from datetime import timedelta
# 自定义监听器
class MyListener(pybreaker.CircuitBreakerListener):
    def state_change(self, breaker, old_state, new_state):
        logging.info(f"State changed from {old_state} to {new_state}")
        if new_state == pybreaker.CircuitBreaker.STATE_OPEN:
            # 发送告警
            send_alert(f"Circuit breaker opened for {breaker.name}")
# 创建熔断器实例
breaker = pybreaker.CircuitBreaker(
    name="api_circuit_breaker",
    fail_max=5,
    reset_timeout=60,
    close_timeout=10,  # 从半开到完全关闭的超时时间
    warm_up_interval=timedelta(seconds=30),  # 预热时间
    listeners=[MyListener()]  # 添加监听器
)
# 完整的调用示例
def safe_api_call(url, fallback_response=None):
    try:
        with breaker:
            result = call_external_service(url)
            return result
    except pybreaker.CircuitBreakerError:
        return fallback_response or {"status": "fallback", "data": None}
    except Exception as e:
        # 记录异常但继续抛出
        logging.error(f"API call failed: {e}")
        raise

使用 `circuitbreaker` 库

pip install circuitbreaker

from circuitbreaker import circuitbreaker
import time
@circuitbreaker(
    failure_threshold=3,       # 失败阈值
    recovery_timeout=30,       # 恢复超时时间
    expected_exception=Exception  # 需要监控的异常类型
)
def call_external_service():
    # 调用外部服务
    response = make_http_request()
    return response
# 自定义配置
@circuitbreaker(
    name="my_service",
    failure_threshold=5,
    recovery_timeout=60,
    expected_exception=(ConnectionError, TimeoutError)
)
def my_unstable_service():
    # 不稳定服务的调用
    pass

手动实现简单的熔断器

如果不想引入第三方库,可以自己实现一个简单的熔断器：

import time
import threading
from enum import Enum
class CircuitState(Enum):
    CLOSED = "closed"      # 正常状态
    OPEN = "open"          # 熔断状态
    HALF_OPEN = "half_open"  # 半开状态
class SimpleCircuitBreaker:
    def __init__(self, failure_threshold=3, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.lock = threading.Lock()
    def call(self, func, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                else:
                    raise Exception("Circuit breaker is open")
        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise e
    def on_success(self):
        with self.lock:
            self.failure_count = 0
            self.state = CircuitState.CLOSED
    def on_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
# 使用示例
breaker = SimpleCircuitBreaker(failure_threshold=3, recovery_timeout=30)
def risky_call():
    # 可能失败的操作
    return 1 / 0
try:
    result = breaker.call(risky_call)
except Exception as e:
    # 降级处理
    result = get_fallback_data()

结合重试策略

在实际应用中,熔断器通常与重试策略结合使用：

import pybreaker
from tenacity import retry, stop_after_attempt, wait_exponential
import logging
logger = logging.getLogger(__name__)
class ServiceWithRetryCircuitBreaker:
    def __init__(self, service_name):
        self.breaker = pybreaker.CircuitBreaker(
            name=service_name,
            fail_max=5,
            reset_timeout=60,
            listeners=[self.create_listener()]
        )
    def create_listener(self):
        class LoggingListener(pybreaker.CircuitBreakerListener):
            def state_change(self, breaker, old_state, new_state):
                logger.info(f"Circuit breaker '{breaker.name}' state: {old_state} -> {new_state}")
        return LoggingListener()
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        reraise=True
    )
    def call_with_retry(self, func, *args, **kwargs):
        with self.breaker:
            return func(*args, **kwargs)
    def safe_call(self, func, fallback_func=None, *args, **kwargs):
        try:
            return self.call_with_retry(func, *args, **kwargs)
        except pybreaker.CircuitBreakerError:
            return fallback_func() if fallback_func else None
        except Exception as e:
            logger.error(f"Service call failed: {e}")
            raise
# 使用
service = ServiceWithRetryCircuitBreaker("my_service")
def fetch_data():
    # 外部API调用
    return requests.get("https://api.example.com/data")
def fallback():
    return {"status": "degraded", "data": cache.get("data")}
result = service.safe_call(fetch_data, fallback)

配置化管理

在大型项目中,推荐使用配置中心管理熔断器参数：

import yaml
import pybreaker
class CircuitBreakerFactory:
    def __init__(self, config_path):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)
        self.breakers = {}
    def get_breaker(self, service_name):
        if service_name not in self.breakers:
            service_config = self.config.get('circuit_breakers', {}).get(service_name, {})
            self.breakers[service_name] = pybreaker.CircuitBreaker(
                name=service_name,
                fail_max=service_config.get('fail_max', 5),
                reset_timeout=service_config.get('reset_timeout', 30)
            )
        return self.breakers[service_name]
# config.yaml
"""
circuit_breakers:
  payment_service:
    fail_max: 3
    reset_timeout: 60
  notification_service:
    fail_max: 5
    reset_timeout: 30
"""
factory = CircuitBreakerFactory("config.yaml")
payment_breaker = factory.get_breaker("payment_service")