Khi tôi bắt đầu làm việc với API, có một lần hệ thống của tôi bị chặn hoàn toàn vì gửi quá nhiều yêu cầu cùng lúc. Sau 3 ngày debug liên tục và mất 200 USD tiền phí phát sinh do retry không kiểm soát, tôi mới hiểu tại sao ba khái niệm này lại quan trọng đến vậy. Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến để bạn tránh những sai lầm tương tự.
Rate Limiting là gì và tại sao cần thiết?
Rate Limiting (giới hạn tốc độ) giống như một người gác cổng kiểm soát số lượng xe được phép vào bãi đỗ mỗi phút. Khi bạn gửi quá nhiều yêu cầu đến API trong thời gian ngắn, server sẽ trả về lỗi 429 (Too Many Requests). Điều này bảo vệ cả bạn lẫn nhà cung cấp dịch vụ.
Cách hoạt động cơ bản
Ví dụ, HolySheheep AI có giới hạn 1000 yêu cầu/phút cho gói Standard. Khi bạn gửi yêu cầu thứ 1001 trong cùng phút đó, API sẽ trả về:
{
"error": {
"message": "Rate limit exceeded",
"type": "rate_limit_error",
"code": 429
}
}Triển khai Rate Limiter đơn giản với Python
import time
import threading
from collections import deque
class SimpleRateLimiter:
"""
Rate Limiter đơn giản theo sliding window
Tự tay implement để hiểu rõ nguyên lý hoạt động
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque()
self.lock = threading.Lock()
def acquire(self) -> bool:
"""
Kiểm tra xem có được phép gửi request không
Returns True nếu được phép, False nếu bị giới hạn
"""
current_time = time.time()
with self.lock:
# Xóa các request cũ đã hết hạn
while self.requests and self.requests[0] < current_time - self.window_seconds:
self.requests.popleft()
# Kiểm tra số lượng request trong window
if len(self.requests) < self.max_requests:
self.requests.append(current_time)
return True
return False
def wait_and_acquire(self):
"""
Chờ đến khi được phép gửi request
Dùng trong batch processing
"""
while not self.acquire():
time.sleep(0.1) # Chờ 100ms rồi thử lại
return True
Sử dụng
limiter = SimpleRateLimiter(max_requests=60, window_seconds=60)
Kiểm tra trước khi gọi API
if limiter.acquire():
# Gọi API ở đây
print("Request được phép")
else:
print("Bị giới hạn, thử lại sau")Retry Strategy - Khi nào và làm thế nào để thử lại?
Retry (thử lại) là cách xử lý khi yêu cầu thất bại do lỗi tạm thời như mạng chậm hoặc server bận. Tuy nhiên, retry không đúng cách có thể gây ra hiệu tượng "retry storm" - hàng ngàn request thử lại cùng lúc làm server quá tải hoàn toàn.
Mô hình Exponential Backoff
Thay vì thử lại ngay lập tức, chúng ta tăng khoảng thời gian chờ theo cấp số nhân. Đây là công thức tôi dùng thực tế:
import random
import time
def calculate_backoff(attempt: int, base_delay: float = 1.0, max_delay: float = 32.0) -> float:
"""
Tính toán thời gian chờ theo Exponential Backoff với Jitter
Công thức: min(max_delay, base_delay * 2^attempt) + random_jitter
- attempt: số lần thử (bắt đầu từ 0)
- base_delay: thời gian chờ cơ bản (giây)
- max_delay: thời gian chờ tối đa (giây)
Jitter ngẫu nhiên giúp tránh retry storm
"""
# Exponential backoff cơ bản
exponential_delay = base_delay * (2 ** attempt)
# Giới hạn thời gian chờ tối đa
capped_delay = min(exponential_delay, max_delay)
# Thêm jitter ngẫu nhiên (0-25% của delay)
jitter = random.uniform(0, capped_delay * 0.25)
return capped_delay + jitter
Ví dụ thời gian chờ
print("Thời gian chờ cho các lần thử:")
for i in range(6):
delay = calculate_backoff(i)
print(f" Lần thử {i}: {delay:.2f} giây")
Output mẫu:
Lần thử 0: 1.15 giây
Lần thử 1: 2.08 giây
Lần thử 2: 4.22 giây
Lần thử 3: 8.01 giây
Lần thử 4: 16.33 giây
Lần thử 5: 32.00 giây
Hàm Retry có khả năng cấu hình
import functools
import time
import logging
logger = logging.getLogger(__name__)
def retry_with_backoff(
max_attempts: int = 3,
base_delay: float = 1.0,
max_delay: float = 32.0,
retryable_errors: tuple = (ConnectionError, TimeoutError)
):
"""
Decorator cho retry với exponential backoff
Ví dụ sử dụng:
@retry_with_backoff(max_attempts=5, base_delay=0.5)
def call_api():
return requests.get("https://api.holysheep.ai/v1/models")
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except retryable_errors as e:
last_exception = e
if attempt == max_attempts - 1:
logger.error(f"Tất cả {max_attempts} lần thử đều thất bại")
raise
delay = calculate_backoff(attempt, base_delay, max_delay)
logger.warning(
f"Lần thử {attempt + 1}/{max_attempts} thất bại: {e}. "
f"Thử lại sau {delay:.2f}s"
)
time.sleep(delay)
raise last_exception
return wrapper
return decorator
Sử dụng thực tế
@retry_with_backoff(max_attempts=4, base_delay=0.5)
def analyze_image_with_retry(image_url: str):
"""
Gọi API phân tích hình ảnh với retry tự động
"""
import requests
response = requests.post(
"https://api.holysheep.ai/v1/image_analysis",
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
json={"image_url": image_url},
timeout=30
)
if response.status_code == 429:
raise ConnectionError("Rate limited")
response.raise_for_status()
return response.json()Degradation Strategy - Giảm tải thông minh
Khi API gặp sự cố hoàn toàn hoặc phản hồi quá chậm, Degradation (giảm tải) giúp hệ thống của bạn tiếp tục hoạt động ở mức hạn chế thay vì hoàn toàn ngừng trệ. Đây là chiến lược "graceful degradation" - giảm chất lượng nhưng vẫn phục vụ được.
Circuit Breaker Pattern
from enum import Enum
from datetime import datetime, timedelta
import threading
class CircuitState(Enum):
CLOSED = "closed" # Bình thường, request đi qua
OPEN = "open" # Lỗi liên tục, ngắt mạch
HALF_OPEN = "half_open" # Thử phục hồi
class CircuitBreaker:
"""
Circuit Breaker ngăn chặn cascade failure
Trạng thái:
- CLOSED: Mọi request đều đi qua, theo dõi lỗi
- OPEN: Từ chối tất cả request, trả về fallback
- HALF_OPEN: Cho phép một số request thử, quyết định phục hồi
"""
def __init__(
self,
failure_threshold: int = 5, # Số lỗi để mở circuit
recovery_timeout: int = 60, # Giây trước khi thử phục hồi
success_threshold: int = 2, # Lỗi thành công để đóng circuit
half_open_max_calls: int = 3 # Số call trong half-open
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.success_threshold = success_threshold
self.half_open_max_calls = half_open_max_calls
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
self._last_failure_time = None
self._half_open_calls = 0
self._lock = threading.Lock()
@property
def state(self) -> CircuitState:
with self._lock:
if self._state == CircuitState.OPEN:
# Kiểm tra timeout để chuyển sang half-open
if (datetime.now() - self._last_failure_time).seconds >= self.recovery_timeout:
self._state = CircuitState.HALF_OPEN
self._half_open_calls = 0
return self._state
def can_execute(self) -> bool:
"""Kiểm tra request có được phép thực thi không"""
state = self.state
if state == CircuitState.CLOSED:
return True
if state == CircuitState.OPEN:
return False
# HALF_OPEN: cho phép một số request thử
with self._lock:
if self._half_open_calls < self.half_open_max_calls:
self._half_open_calls += 1
return True
return False
def record_success(self):
"""Ghi nhận request thành công"""
with self._lock:
if self._state == CircuitState.HALF_OPEN:
self._success_count += 1
if self._success_count >= self.success_threshold:
self._state = CircuitState.CLOSED
self._failure_count = 0
self._success_count = 0
else:
self._failure_count = 0
def record_failure(self):
"""Ghi nhận request thất bại"""
with self._lock:
self._failure_count += 1
self._last_failure_time = datetime.now()
if self._state == CircuitState.HALF_OPEN:
# Thất bại trong half-open -> quay lại open
self._state = CircuitState.OPEN
self._success_count = 0
elif self._failure_count >= self.failure_threshold:
self._state = CircuitState.OPEN
Sử dụng
circuit = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
def call_api_with_circuit(image_data: bytes):
"""Gọi API với circuit breaker"""
if not circuit.can_execute():
print("⚠️ Circuit OPEN - Sử dụng fallback")
return get_fallback_response()
try:
# Gọi API thực tế
result = call_holysheep_api(image_data)
circuit.record_success()
return result
except Exception as e:
circuit.record_failure()
return get_fallback_response()
def get_fallback_response():
"""Trả về response mặc định khi API lỗi"""
return {
"status": "degraded",
"message": "Service temporarily degraded",
"fallback": True
}Kết hợp cả ba chiến lược
Trong thực tế, tôi thường kết hợp cả ba chiến lược này vào một lớp xử lý duy nhất. Dưới đây là implementation hoàn chỉnh:
import time
import threading
from typing import Optional, Callable, Any
from dataclasses import dataclass
@dataclass
class APIConfig:
"""Cấu hình cho API client"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
rate_limit: int = 60 # requests per window
rate_window: int = 60 # seconds
max_retries: int = 3
timeout: int = 30
circuit_threshold: int = 5 # failures before opening
circuit_recovery: int = 60 # seconds before trying again
class ResilientAPIClient:
"""
API Client với đầy đủ chiến lược resilience:
- Rate Limiting
- Retry với Exponential Backoff
- Circuit Breaker
- Fallback Response
"""
def __init__(self, config: APIConfig):
self.config = config
self.rate_limiter = SimpleRateLimiter(config.rate_limit, config.rate_window)
self.circuit_breaker = CircuitBreaker(
failure_threshold=config.circuit_threshold,
recovery_timeout=config.circuit_recovery
)
self._request_count = 0
self._total_latency = 0.0
self._lock = threading.Lock()
def call(
self,
endpoint: str,
method: str = "POST",
data: Optional[dict] = None,
fallback: Optional[Callable] = None
) -> dict:
"""
Gọi API với tất cả resilience strategies
Args:
endpoint: API endpoint (vd: "/chat/completions")
method: HTTP method
data: Request body
fallback: Hàm fallback nếu cần
Returns:
Response dict hoặc fallback response
"""
# 1. Kiểm tra Circuit Breaker
if not self.circuit_breaker.can_execute():
print("🔴 Circuit Breaker OPEN - Trả về fallback")
return fallback() if fallback else {"error": "service_unavailable"}
# 2. Kiểm tra Rate Limit
if not self.rate_limiter.acquire():
wait_time = 0.1
while not self.rate_limiter.acquire():
time.sleep(wait_time)
wait_time = min(wait_time * 1.5, 2.0) # Tăng dần, max 2s
# 3. Thực hiện request với retry
for attempt in range(self.config.max_retries):
try:
start_time = time.time()
# Gọi API thực tế
response = self._execute_request(endpoint, method, data)
# Ghi nhận metrics
latency = time.time() - start_time
self._record_metrics(response.status_code, latency)
self.circuit_breaker.record_success()
return response.json()
except Exception as e:
if attempt == self.config.max_retries - 1:
self.circuit_breaker.record_failure()
return fallback() if fallback else {"error": str(e)}
delay = calculate_backoff(attempt, base_delay=0.5)
print(f"⚡ Retry {attempt + 1} sau {delay:.2f}s: {e}")
time.sleep(delay)
return fallback() if fallback else {"error": "max_retries_exceeded"}
def _execute_request(self, endpoint: str, method: str, data: dict) -> Any:
"""Thực hiện HTTP request"""
import requests
url = f"{self.config.base_url}{endpoint}"
headers = {
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
}
response = requests.post(
url,
headers=headers,
json=data,
timeout=self.config.timeout
)
if response.status_code