ผมเคยเจอสถานการณ์ที่ทำให้หัวใจหยุดเต้นชั่วขณะ — ระบบแชทบอทที่พัฒนาด้วย Python กระทันหันหยุดทำงานกลางคัน เมื่อเซิร์ฟเวอร์ AI API ตอบสนองช้าผิดปกติ และผู้ใช้งานกว่า 5,000 รายต่างเห็นข้อความ ConnectionError: timeout after 30 seconds พร้อมกัน หลังจากวิเคราะห์ logs พบว่าปัญหาเกิดจากการไม่มีกลไกจัดการความล้มเหลวที่เหมาะสม บทความนี้จะพาคุณสร้าง API Gateway ที่มี Circuit Breaker และ Retry Mechanism อย่างมืออาชีพ โดยใช้ HolySheep AI เป็นตัวอย่างในการเรียกใช้งาน
ทำไมต้องมี Circuit Breaker และ Retry?
ในระบบ Microservices หรือการเรียกใช้ External API ทุกอย่างมีความเสี่ยงที่จะล้มเหลว ไม่ว่าจะเป็น:
- Network timeout หรือ packet loss
- API server ล่มชั่วคราว
- Rate limit exceeded
- Invalid credentials เช่น 401 Unauthorized
- Service overload ทำให้ response ช้าผิดปกติ
เมื่อไม่มีกลไกป้องกัน ระบบจะตกอยู่ในสถานะ Cascade Failure คือ request ที่รอคอยจะค่อยๆ สะสมจนใช้ทรัพยากรจนหมด (Thread pool exhaustion) และระบบทั้งหมดล่มในที่สุด การใช้ Circuit Breaker จะช่วยตัดวงจรการเรียกเมื่อพบว่า API มีปัญหาต่อเนื่อง ส่วน Retry Mechanism จะช่วยจัดการกับความล้มเหลวชั่วคราว (transient failures)
การสร้าง Circuit Breaker พื้นฐาน
มาเริ่มต้นด้วยการสร้าง Circuit Breaker class ที่จะควบคุมการเรียก API แต่ละครั้ง หลักการทำงานคือมี 3 สถานะหลัก: CLOSED (ปกติ), OPEN (หยุดเรียกชั่วคราว), และ HALF_OPEN (ทดสอบว่าฟื้นตัวได้หรือยัง)
import time
import threading
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
from functools import wraps
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5
success_threshold: int = 2
timeout: float = 30.0
expected_exceptions: tuple = (Exception,)
@dataclass
class CircuitBreakerStats:
failure_count: int = 0
success_count: int = 0
last_failure_time: Optional[float] = None
state: CircuitState = CircuitState.CLOSED
total_calls: int = 0
total_failures: int = 0
class CircuitBreaker:
def __init__(self, config: CircuitBreakerConfig):
self.config = config
self.stats = CircuitBreakerStats()
self._lock = threading.RLock()
@property
def state(self) -> CircuitState:
with self._lock:
if self.stats.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.stats.state = CircuitState.HALF_OPEN
logger.info("Circuit changed: OPEN -> HALF_OPEN (timeout expired)")
return self.stats.state
def _should_attempt_reset(self) -> bool:
if self.stats.last_failure_time is None:
return False
return (time.time() - self.stats.last_failure_time) >= self.config.timeout
def record_success(self):
with self._lock:
self.stats.total_calls += 1
self.stats.success_count += 1
self.stats.last_failure_time = None
if self.stats.state == CircuitState.HALF_OPEN:
if self.stats.success_count >= self.config.success_threshold:
self.stats.state = CircuitState.CLOSED
self.stats.failure_count = 0
self.stats.success_count = 0
logger.info("Circuit changed: HALF_OPEN -> CLOSED (recovered)")
def record_failure(self):
with self._lock:
self.stats.total_calls += 1
self.stats.total_failures += 1
self.stats.failure_count += 1
self.stats.last_failure_time = time.time()
if self.stats.state == CircuitState.HALF_OPEN:
self.stats.state = CircuitState.OPEN
self.stats.success_count = 0
logger.warning("Circuit changed: HALF_OPEN -> OPEN (failure during recovery)")
elif (self.stats.state == Circuit