带退避队列的断路器：消息 enrichment 失败退避队列处理方案咨询

Great question—let’s break this down clearly, since this is a common pattern for building resilient message processing pipelines.

To clarify: A standard circuit breaker stops direct calls to a failing external service to prevent cascading failures, but it doesn’t handle what happens to the requests/messages that would have been sent to that service. The "with back-out queue" extension solves exactly this problem: when the circuit breaker trips (opens), instead of dropping or rejecting the message, you persist it to a dedicated queue for later retry. This aligns perfectly with your requirement:

• External resource is intermittently unavailable → trigger circuit breaker open
• Route failed messages to a back-out queue
• Poll the queue later to reprocess once the resource recovers

Let’s break down the core components and key logic you’ll need:

2.1 Core Component Design #

• Circuit Breaker: Monitors the health of your external resources. It should track:
  • State: Closed (normal operation, allow calls), Open (block direct calls, route to back-out queue), Half-Open (test if the resource has recovered with limited calls)
  • Thresholds: Define when to trip the breaker (e.g., 3 consecutive failures, or 80% failure rate in 1 minute)
• Back-Out Queue: A persistent queue (use something like RabbitMQ, Kafka, or even a database table) to store messages that couldn’t be processed due to external failures. Each message should include metadata like:
  • Original message content
  • Retry count
  • Last failure timestamp
  • Failure reason (e.g., "timeout", "connection refused")
• Retry Processor: A scheduled job that polls the back-out queue and attempts reprocessing. It should follow a retry strategy to avoid overwhelming recovering services.
• Main Message Router: The entry point for incoming messages. It checks the circuit breaker state first:
  • If breaker is closed: Proceed with enrichment and business logic
  • If breaker is open: Send the message directly to the back-out queue

2.2 Key Implementation Details #

Circuit Breaker State Logic #

• Closed → Open: When the failure threshold is hit, switch to open state and start a timer. After a set timeout (e.g., 5 minutes), switch to half-open to test recovery.
• Half-Open → Closed: If a test call to the external resource succeeds, reset the failure count and switch back to closed. If it fails, revert to open state.
• Half-Open → Open: If the test call fails, go back to open and reset the timer.

Retry Strategy #

Use exponential backoff with a maximum retry limit to balance retry speed and resource load:

• Example: 1 minute wait for first retry, 2 minutes for second, 4 minutes for third, up to 5 total retries. After that, move the message to a dead-letter queue for manual review.

Idempotency Guarantee #

Since messages will be retried, your business logic must be idempotent. Use a unique message ID to check if the message has already been processed before executing business actions (e.g., look up the ID in a processed messages table).

2.3 Simplified Pseudocode Example #

import time

class CircuitBreaker:
    def __init__(self, failure_threshold=3, open_timeout=300):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.open_timeout = open_timeout  # 5 minutes in seconds
        self.state = "closed"
        self.opened_at = None

    def record_success(self):
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            self.opened_at = time.time()

    def can_make_request(self):
        if self.state == "open":
            # Check if it's time to test recovery
            if time.time() - self.opened_at >= self.open_timeout:
                self.state = "half-open"
                return True
            return False
        return True

# Main message processing flow
def handle_incoming_message(raw_message):
    breaker = CircuitBreaker()  # In practice, use a singleton per external resource
    if not breaker.can_make_request():
        # Send to back-out queue with metadata
        back_out_queue.enqueue({
            "original_message": raw_message,
            "retry_count": 0,
            "last_failed_at": time.time(),
            "failure_reason": "Circuit breaker is open"
        })
        return

    try:
        # Execute enrichment via external resources
        enriched_msg = call_external_enrichment(raw_message)
        # Run downstream business logic
        execute_business_workflow(enriched_msg)
        breaker.record_success()
    except ExternalResourceUnavailableError as e:
        breaker.record_failure()
        # Send failed message to back-out queue
        back_out_queue.enqueue({
            "original_message": raw_message,
            "retry_count": 0,
            "last_failed_at": time.time(),
            "failure_reason": str(e)
        })

# Scheduled retry processor
def process_back_out_queue():
    while True:
        # Poll queue for messages ready to retry
        pending_messages = back_out_queue.get_pending()
        for msg in pending_messages:
            max_retries = 5
            if msg["retry_count"] >= max_retries:
                # Move to dead-letter queue for manual handling
                dead_letter_queue.enqueue(msg)
                back_out_queue.remove(msg)
                continue

            # Calculate exponential backoff wait time
            wait_time = (2 ** msg["retry_count"]) * 60  # Minutes to seconds
            if time.time() - msg["last_failed_at"] < wait_time:
                # Not ready to retry yet, skip
                continue

            try:
                # Attempt reprocessing
                enriched_msg = call_external_enrichment(msg["original_message"])
                execute_business_workflow(enriched_msg)
                back_out_queue.remove(msg)
            except ExternalResourceUnavailableError:
                # Update retry metadata and put back in queue
                msg["retry_count"] += 1
                msg["last_failed_at"] = time.time()
                back_out_queue.update(msg)
        # Poll every minute
        time.sleep(60)

• Per-Resource Breakers: If you’re calling multiple external resources, use a separate circuit breaker for each. This prevents one failing resource from taking down your entire pipeline.
• Monitoring & Alerts: Track breaker state changes, back-out queue size, and retry success rates. Alert your team if the breaker stays open for too long or the queue starts piling up.
• Queue Persistence: Ensure your back-out queue is persistent (e.g., enable disk persistence in RabbitMQ/Kafka) to avoid losing messages if your service restarts.

内容的提问来源于stack exchange，提问作者Darius X.