Khi thực tế không như là mơ
Ba giờ sáng thứ Hai, điện thoại mình rung liên hồi. Con bot crawl dữ liệu vốn đang chạy êm ru bỗng dưng “lăn đùng” ra chết. Kiểm tra log thì chỉ thấy vỏn vẹn dòng lỗi ConnectionTimeout. Ở môi trường local, mọi thứ hoàn hảo 100%. Thế nhưng khi đẩy lên server, cứ trung bình 3 lần/tuần, nó lại dở chứng vì API đối tác phản hồi chậm.
Dự án này ban đầu chỉ có 200 dòng code. Sau 3 tháng vận hành, nó phình lên hơn 2.000 dòng. Đáng buồn là một nửa trong số đó không phải logic nghiệp vụ. Đó là những đoạn code chắp vá để xử lý mạng chập chờn. Đây là bài học đắt giá về xây dựng hệ thống bền bỉ (resilient system) mà mọi developer đều phải trải qua.
Tại sao code của bạn hay “chết yểu”?
Trong thế giới hệ thống phân tán, chúng ta đối mặt với Transient Faults (Lỗi tạm thời). Đây là những lỗi tự đến rồi tự đi. Nếu bạn thử lại (retry) sau vài giây, khả năng cao là mọi thứ lại chạy mượt. Các tình huống kinh điển bao gồm:
- Mất kết nối Internet trong tích tắc (micro-outages).
- Server đối tác quá tải và trả về lỗi 429 (Rate limiting).
- Database tạm khóa để thực hiện backup định kỳ.
- Microservices phản hồi chậm do nghẽn băng thông.
Viết code kiểu “happy path” là sai lầm chết người. Chỉ cần mạng lag 1-2 giây, toàn bộ quy trình phía sau sẽ sụp đổ như quân bài domino. Bạn không thể kiểm soát được hạ tầng mạng, nhưng bạn có thể kiểm soát cách code phản ứng với nó.
Những cách xử lý lỗi “cồng kềnh” thường gặp
Cách 1: Vòng lặp While và Try-Except (Code Smell)
Đây là cách làm bản năng nhất. Hầu như ai mới học Python cũng từng viết như thế này:
import time
attempts = 0
while attempts < 3:
try:
result = call_api_service()
break
except Exception as e:
attempts += 1
print(f"Thử lại lần {attempts}...")
time.sleep(2)
else:
print("Thất bại hoàn toàn.")Code này chạy được nhưng rất “bẩn”. Thử tưởng tượng bạn có 50 endpoint cần gọi. File code sẽ ngập ngụa trong vòng lặp while và biến đếm. Nó làm loãng logic chính, cực kỳ khó bảo trì và rất dễ gây ra lỗi logic như quên tăng biến đếm dẫn đến vòng lặp vô tận.
Cách 2: Tự viết Decorator (Vẫn chưa tối ưu)
Khá hơn một chút, bạn có thể tự viết decorator để tái sử dụng. Tuy nhiên, việc tự xử lý các bài toán như Exponential Backoff (chờ tăng dần) hay lọc đúng loại Exception để retry là một cực hình. Đừng tốn công viết lại những thứ mà cộng đồng đã tối ưu hóa suốt nhiều năm.
Giải pháp chuyên nghiệp: Thư viện Tenacity
Tenacity là thư viện Python sinh ra để giải quyết bài toán retry một cách thanh lịch. Thay vì viết code điều khiển (imperative), bạn chỉ cần khai báo (declarative) mong muốn của mình thông qua Decorators.
Cài đặt nhanh
pip install tenacity1. Cơ chế Retry cơ bản
Gắn @retry lên đầu hàm và xong! Mặc định, nó sẽ thử lại liên tục cho đến khi hàm chạy thành công mới thôi.
from tenacity import retry
@retry
def do_something_unreliable():
print("Đang cố kết nối...")
raise Exception("Lỗi mạng!")2. Kiểm soát thời gian và số lần thử
Trong thực tế, chúng ta không thể đợi mãi. Bạn cần đặt ra giới hạn chịu đựng cho ứng dụng.
from tenacity import retry, stop_after_attempt, stop_after_delay
# Chỉ thử tối đa 5 lần
@retry(stop=stop_after_attempt(5))
def call_api():
raise IOError("Fail")
# Dừng lại sau 10 giây bất kể đã thử bao nhiêu lần
@retry(stop=stop_after_delay(10))
def connect_db():
pass3. Chiến thuật chờ thông minh (Wait strategies)
Đừng retry ngay lập tức! Nếu server đang quá tải mà bạn nã request liên tục, bạn sẽ bị chặn IP ngay lập tức. Hãy dùng Exponential Backoff.
from tenacity import retry, wait_exponential
# Lần 1 chờ 1s, lần 2 chờ 2s, lần 3 chờ 4s... tối đa 10s
@retry(wait=wait_exponential(multiplier=1, min=1, max=10))
def fetch_data():
raise Exception("Server bận")4. Chỉ Retry khi thực sự cần thiết
Nếu gặp lỗi 404 Not Found, bạn có thử lại 1.000 lần cũng vô ích. Tenacity cho phép bạn lọc chính xác loại lỗi cần retry.
import requests
from tenacity import retry, retry_if_exception_type
@retry(retry=retry_if_exception_type(requests.exceptions.ConnectionError))
def get_weather():
return requests.get("https://api.weather.com/v1/...")Áp dụng thực tế: Xây dựng hàm gọi API chuẩn chỉnh
Đây là mẫu code mình thường dùng cho các dự án gọi OpenAI API hoặc thanh toán điện tử. Nó kết hợp logging để bạn dễ dàng theo dõi hệ thống đang làm gì đằng sau hậu trường.
import logging
import requests
from tenacity import retry, stop_after_attempt, wait_fixed, before_sleep_log, retry_if_exception
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def is_transient_error(exception):
if isinstance(exception, requests.exceptions.HTTPError):
# Chỉ retry nếu là lỗi 429 (Rate limit) hoặc lỗi server 5xx
return exception.response.status_code in [429, 500, 502, 503, 504]
return isinstance(exception, (requests.exceptions.ConnectionError, requests.exceptions.Timeout))
@retry(
stop=stop_after_attempt(3),
wait=wait_fixed(2),
before_sleep=before_sleep_log(logger, logging.INFO),
retry=retry_if_exception(is_transient_error)
)
def safe_get_data(url):
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()
try:
data = safe_get_data("https://api.example.com/data")
print("Lấy dữ liệu thành công!")
except Exception as e:
print(f"Thất bại sau 3 lần thử: {e}")Lời kết
Sự khác biệt giữa một Senior và Junior nằm ở cách họ xử lý những tình huống không hoàn hảo. Sử dụng Tenacity không chỉ giúp code của bạn sạch hơn (Clean Code) mà còn giúp hệ thống tự phục hồi kỳ diệu. Hãy bắt đầu áp dụng nó vào dự án ngay hôm nay. Tin mình đi, bạn sẽ ngủ ngon hơn vì không còn những cuộc gọi báo lỗi lúc nửa đêm!
