IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python脚本熔断器模式如何写

wen 实用脚本 3

本文目录导读：

Python脚本熔断器模式如何写

  1. 基础熔断器实现
  2. 装饰器版本
  3. 高级版本（支持异步）
  4. 使用第三方库（pybreaker）
  5. 关键设计要点

我来详细说明Python中熔断器模式的实现方式。

基础熔断器实现

import time
from enum import Enum
from functools import wraps
class CircuitBreakerState(Enum):
    CLOSED = "closed"      # 正常状态
    OPEN = "open"          # 熔断状态
    HALF_OPEN = "half_open" # 半开状态
class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30, half_open_max_calls=3):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        self.state = CircuitBreakerState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0
    def call(self, func, *args, **kwargs):
        if self.state == CircuitBreakerState.OPEN:
            if self._should_attempt_recovery():
                self.state = CircuitBreakerState.HALF_OPEN
                self.half_open_calls = 0
            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):
        if self.state == CircuitBreakerState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                self.state = CircuitBreakerState.CLOSED
                self.failure_count = 0
        elif self.state == CircuitBreakerState.CLOSED:
            self.failure_count = 0
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.state == CircuitBreakerState.HALF_OPEN:
            self.state = CircuitBreakerState.OPEN
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitBreakerState.OPEN
    def _should_attempt_recovery(self):
        if self.last_failure_time is None:
            return True
        return time.time() - self.last_failure_time >= self.recovery_timeout
# 使用示例
def unreliable_service():
    import random
    if random.random() < 0.7:  # 70% 概率失败
        raise Exception("Service failed!")
    return "Success"
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=10)
for i in range(20):
    try:
        result = cb.call(unreliable_service)
        print(f"Call {i}: {result}")
    except Exception as e:
        print(f"Call {i}: {e}")
    time.sleep(1)

装饰器版本

import time
from functools import wraps
from enum import Enum
class CircuitBreakerDecorator:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = "closed"
        self.failure_count = 0
        self.last_failure_time = None
    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if self.state == "open":
                if time.time() - self.last_failure_time >= self.recovery_timeout:
                    self.state = "half_open"
                    print("Attempting recovery...")
                else:
                    raise Exception(f"Circuit breaker is OPEN for {func.__name__}")
            try:
                result = func(*args, **kwargs)
                self._handle_success()
                return result
            except Exception as e:
                self._handle_failure()
                raise e
        return wrapper
    def _handle_success(self):
        if self.state == "half_open":
            self.state = "closed"
            self.failure_count = 0
            print("Service restored, circuit closed")
        elif self.state == "closed":
            self.failure_count = 0
    def _handle_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.state == "half_open" or self.failure_count >= self.failure_threshold:
            self.state = "open"
            print(f"Circuit OPEN after {self.failure_count} failures")
# 使用装饰器
@CircuitBreakerDecorator(failure_threshold=3, recovery_timeout=5)
def api_call():
    import random
    if random.random() < 0.6:
        raise Exception("API call failed")
    return "API call succeeded"
# 测试
for i in range(15):
    try:
        result = api_call()
        print(f"Call {i}: {result}")
    except Exception as e:
        print(f"Call {i}: {str(e)}")
    time.sleep(1)

高级版本（支持异步）

import asyncio
import time
from enum import Enum
from functools import wraps
from typing import Callable, Any, Optional
class CircuitBreakerState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"
class AdvancedCircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_max_calls: int = 3,
        excluded_exceptions: Optional[tuple] = None
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        self.excluded_exceptions = excluded_exceptions or ()
        self.state = CircuitBreakerState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self._lock = asyncio.Lock()
    async def __aenter__(self):
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        pass
    async def call_async(self, func: Callable, *args, **kwargs) -> Any:
        async with self._lock:
            await self._check_state()
        try:
            if asyncio.iscoroutinefunction(func):
                result = await func(*args, **kwargs)
            else:
                result = func(*args, **kwargs)
            await self._on_success()
            return result
        except self.excluded_exceptions:
            raise
        except Exception as e:
            await self._on_failure()
            raise
    async def _check_state(self):
        if self.state == CircuitBreakerState.OPEN:
            if self._should_attempt_recovery():
                print("Attempting recovery...")
                self.state = CircuitBreakerState.HALF_OPEN
                self.success_count = 0
            else:
                raise Exception("Circuit breaker is OPEN")
    async def _on_success(self):
        if self.state == CircuitBreakerState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.half_open_max_calls:
                self.state = CircuitBreakerState.CLOSED
                self.failure_count = 0
                print("Circuit CLOSED - service restored")
        elif self.state == CircuitBreakerState.CLOSED:
            self.failure_count = 0
    async def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.state == CircuitBreakerState.HALF_OPEN:
            self.state = CircuitBreakerState.OPEN
            print("Circuit OPEN - recovery failed")
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitBreakerState.OPEN
            print(f"Circuit OPEN after {self.failure_count} failures")
    def _should_attempt_recovery(self):
        if self.last_failure_time is None:
            return True
        return time.time() - self.last_failure_time >= self.recovery_timeout
# 异步使用示例
async def async_api_call():
    import random
    await asyncio.sleep(0.1)
    if random.random() < 0.6:
        raise Exception("Async API failed")
    return "Async API succeeded"
async def main():
    cb = AdvancedCircuitBreaker(
        failure_threshold=3,
        recovery_timeout=5,
        half_open_max_calls=2
    )
    for i in range(20):
        try:
            result = await cb.call_async(async_api_call)
            print(f"Call {i}: {result}")
        except Exception as e:
            print(f"Call {i}: {str(e)}")
        await asyncio.sleep(1)
# 运行
if __name__ == "__main__":
    asyncio.run(main())

使用第三方库（pybreaker）

# pip install pybreaker
import pybreaker
import random
import time
# 创建熔断器实例
breaker = pybreaker.CircuitBreaker(
    fail_max=3,       # 最大失败次数
    reset_timeout=10, # 重置超时（秒）
    exclude=[         # 排除的异常类型
        ValueError
    ]
)
# 使用装饰器
@breaker
def unstable_service():
    if random.random() < 0.7:
        raise ConnectionError("Service unreachable")
    return "Service call succeeded"
# 或者使用上下文管理器
def another_service():
    with breaker:
        if random.random() < 0.7:
            raise ConnectionError("Service unreachable")
        return "Another service call succeeded"
# 监听器模式
class MyListener(pybreaker.CircuitBreakerListener):
    def state_change(self, cb, old_state, new_state):
        print(f"State changed from {old_state} to {new_state}")
    def failure(self, cb, exc):
        print(f"Failure detected: {exc}")
    def success(self, cb):
        print("Success detected")
breaker_with_listener = pybreaker.CircuitBreaker(
    fail_max=3,
    reset_timeout=10,
    listeners=[MyListener()]
)
# 测试
for i in range(15):
    try:
        result = unstable_service()
        print(f"Call {i}: {result}")
    except pybreaker.CircuitBreakerError as e:
        print(f"Call {i}: Circuit breaker open")
    except Exception as e:
        print(f"Call {i}: {str(e)}")
    time.sleep(1)

关键设计要点

  1. 状态管理：CLOSED → OPEN → HALF_OPEN 的状态转换
  2. 失败计数：统计连续失败次数
  3. 超时恢复：自动尝试恢复服务
  4. 半开测试：部分允许请求来测试服务是否恢复
  5. 异常处理：可配置排除某些异常

选择哪种实现取决于你的具体需求：

  • 简单场景：使用基础版本
  • 需要装饰器：使用装饰器版本
  • 异步应用：使用异步版本
  • 生产环境：推荐使用成熟的第三方库如 pybreaker

上一篇Python脚本限流算法有哪些实现方式

下一篇Python脚本缓存淘汰策略如何实现

相关文章

抱歉，评论功能暂时关闭!