3 điểm bởi byexist 2026-01-05 | Chưa có bình luận nào. | Chia sẻ qua WhatsApp

Khi sử dụng các thư viện Circuit Breaker Python hiện có (circuitbreaker, pybreaker v.v.), tôi thấy có một số điểm còn thiếu sót.

  • Chỉ dùng bộ đếm lỗi liên tiếp: chỉ cần thành công một lần là bị reset, nên khó phát hiện các dịch vụ không ổn định
  • Khi phục hồi chỉ cho phép một lệnh gọi kiểm tra duy nhất: chưa đủ cho các trường hợp như cache warming

Tôi lấy cảm hứng từ hỗ trợ toán tử của Django Permission và cách quản lý window của resilience4j.

Cài đặt

pip install fluxgate  

Cách dùng cơ bản

from fluxgate import CircuitBreaker  
  
cb = CircuitBreaker(name="external_api")  
  
@cb  
def call_api():  
    return requests.get("https://api.example.com/data";)  

Giá trị mặc định: trip khi 50% thất bại trong 100 lần gọi gần nhất, thử phục hồi sau 60 giây

Hỗ trợ bất đồng bộ

from fluxgate import AsyncCircuitBreaker  
  
cb = AsyncCircuitBreaker(name="external_api")  
  
@cb  
async def call_api():  
    async with httpx.AsyncClient() as client:  
        return await client.get("https://api.example.com/data";)  

Tính năng cốt lõi

Sliding window + trip dựa trên tỷ lệ lỗi

from fluxgate import CircuitBreaker  
from fluxgate.trippers import MinRequests, FailureRate, FailureStreak  
  
cb = CircuitBreaker(  
    name="api",  
    tripper=FailureStreak(5) | (MinRequests(100) & FailureRate(0.5))  
)  

→ Trip khi lỗi liên tiếp 5 lần hoặc khi 50% thất bại trong 100 lệnh gọi gần nhất

Phục hồi tăng dần (RampUp)

from fluxgate.permits import RampUp  
  
cb = CircuitBreaker(  
    name="api",  
    permit=RampUp(initial=0.1, final=1.0, duration=60.0)  
)  

→ Khi phục hồi, lưu lượng tăng dần từ 10% lên 100% trong 60 giây

So sánh tính năng với các thư viện hiện có

Tính năng fluxgate circuitbreaker pybreaker aiobreaker
Sliding window O X X X
Trip theo tỷ lệ lỗi O X X X
Quy tắc có thể kết hợp O X X X
Phục hồi tăng dần O X X X

Liên kết

Rất mong nhận được nhiều phản hồi!

Chưa có bình luận nào.

Chưa có bình luận nào.