Что такое Circuit breaker?

Что такое Circuit breaker

Circuit breaker — это паттерн проектирования, который предотвращает приложение от повторяющихся попыток вызвать неработающий сервис. Его можно назвать «электрическим автоматом» для микросервисов — когда система обнаруживает слишком много ошибок, она прерывает соединение, вместо того чтобы продолжать стучаться в неработающий сервис.

Три состояния Circuit Breaker

1. Closed (нормальное состояние):

# Запросы идут нормально
# Ошибки считаются
# Когда ошибок > порога → переходим в Open

2. Open (цепь разорвана):

# Запросы блокируются моментально
# Не ждём ответа от неработающего сервиса
# Возвращаем ошибку fail-fast
# Через timeout → переходим в Half-Open

3. Half-Open (полуоткрытое):

# Пропускаем несколько пробных запросов
# Если они успешны → возвращаемся в Closed
# Если они падают → возвращаемся в Open

Пример с PyBreaker

from pybreaker import CircuitBreaker
import requests

# Создаём circuit breaker
breaker = CircuitBreaker(
    fail_max=5,              # Порог ошибок: открыть при 5 ошибках
    reset_timeout=60,        # Через 60 сек переходить в Half-Open
    exclude=[TimeoutError]   # Не считать эти ошибки
)

@breaker
def call_external_api(url):
    response = requests.get(url, timeout=5)
    return response.json()

# Использование
try:
    data = call_external_api('https://api.example.com/data')
except:
    print('API недоступен, используем кэш или fallback')

Практический пример с FastAPI

from fastapi import FastAPI, HTTPException
from pybreaker import CircuitBreaker
import httpx
import asyncio

app = FastAPI()

# Circuit breaker для внешнего сервиса
payment_breaker = CircuitBreaker(
    fail_max=3,
    reset_timeout=30,
    name='PaymentService'
)

@app.post('/process-payment/{order_id}')
async def process_payment(order_id: str):
    try:
        # Вызываем защищённый сервис
        result = await call_payment_service(order_id)
        return {'status': 'success', 'data': result}
    except Exception as e:
        raise HTTPException(
            status_code=503,
            detail=f'Payment service unavailable: {str(e)}'
        )

@payment_breaker
async def call_payment_service(order_id: str):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            'https://payment-api.example.com/process',
            json={'order_id': order_id},
            timeout=5
        )
        return response.json()

Custom реализация Circuit Breaker

from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, Any
import asyncio

class CircuitState(Enum):
    CLOSED = 'closed'
    OPEN = 'open'
    HALF_OPEN = 'half_open'

class SimpleCircuitBreaker:
    def __init__(
        self,
        fail_threshold: int = 5,
        success_threshold: int = 2,
        timeout_seconds: int = 60
    ):
        self.fail_threshold = fail_threshold
        self.success_threshold = success_threshold
        self.timeout_seconds = timeout_seconds
        
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
    
    def call(self, func: Callable, *args, **kwargs) -> Any:
        # Проверяем timeout для Half-Open
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
                self.success_count = 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
    
    def _on_success(self):
        self.failure_count = 0
        
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
        elif self.failure_count >= self.fail_threshold:
            self.state = CircuitState.OPEN
    
    def _should_attempt_reset(self) -> bool:
        if not self.last_failure_time:
            return True
        return (
            datetime.now() - self.last_failure_time
        ) > timedelta(seconds=self.timeout_seconds)

# Использование
breaker = SimpleCircuitBreaker(fail_threshold=3)

def external_api_call():
    # Имитация вызова
    return {'status': 'ok'}

try:
    result = breaker.call(external_api_call)
except Exception as e:
    print(f'Сервис недоступен: {e}')

Интеграция с retry логикой

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
import asyncio

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(ConnectionError)
)
async def call_with_retry(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=5)
        return response.json()

# Комбинируем с circuit breaker
breaker = CircuitBreaker(fail_max=3, reset_timeout=30)

@breaker
async def resilient_api_call(url: str):
    return await call_with_retry(url)

Fallback и graceful degradation

class ResilientPaymentService:
    def __init__(self):
        self.breaker = CircuitBreaker(fail_max=3)
        self.cache = {}  # Кэш для fallback
    
    async def process_payment(self, order_id: str):
        try:
            result = await self._call_payment_api(order_id)
            self.cache[order_id] = result
            return result
        except:
            # Fallback: используем кэш
            if order_id in self.cache:
                return self.cache[order_id]
            
            # Или возвращаем заглушку
            return {
                'status': 'pending',
                'message': 'Payment service temporarily unavailable'
            }
    
    @property
    def breaker(self):
        # Публичный доступ к статусу
        return self._breaker
    
    @breaker.setter
    def breaker(self, value):
        self._breaker = value
    
    async def _call_payment_api(self, order_id: str):
        return self.breaker.call(lambda: requests.post(
            'https://payment-api.example.com/process',
            json={'order_id': order_id}
        ).json())

Мониторинг и алерты

from prometheus_client import Counter, Gauge

breaker_state = Gauge(
    'circuit_breaker_state',
    'Состояние circuit breaker (0=closed, 1=open, 2=half_open)',
    ['service']
)

breaker_failures = Counter(
    'circuit_breaker_failures_total',
    'Количество отказов',
    ['service']
)

def monitor_breaker(breaker: CircuitBreaker, service_name: str):
    state_values = {'closed': 0, 'open': 1, 'half_open': 2}
    breaker_state.labels(service=service_name).set(
        state_values[breaker.state.value]
    )

Когда использовать Circuit Breaker

✅ Используй для:

• Вызовов внешних API
• Обращений к БД
• RPC вызовов между микросервисами
• Любых операций, которые могут долго выполняться

❌ НЕ используй для:

• Локальных вычислений
• Быстрых in-memory операций
• Операций без риска таймаутов

Лучшие практики

• Правильно выбирай пороги: слишком низкие = частые срабатывания, слишком высокие = долгая недоступность
• Логируй переходы состояний: CLOSED → OPEN очень важно для алертинга
• Комбинируй с retry и timeout: circuit breaker работает лучше с другими паттернами
• Используй разные breaker'ы для разных сервисов: не создавай один на всех
• Мониторь метрики: отслеживай failure rate и время восстановления

Circuit breaker защищает систему от каскадного отказа и позволяет приложению быстро восстановиться после сбоев в зависимостях.