# Message Queue Design > Implements production-grade message queue design patterns (delivery semantics, ordering guarantees, consumer groups, dead letter queues, priority and delayed delivery) for reliable distributed systems. - Skill: `paulpas/message-queue-design` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/message-queue-design` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/message-queue-design/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/message-queue-design --- # Message Queue Design Patterns Implements production-grade message queue design patterns for reliable distributed systems. Covers delivery semantics (at-least-once, at-most-once, exactly-once), ordering guarantees, consumer group management, dead letter queues, priority and delayed delivery, and broker selection decision frameworks to match infrastructure to workload requirements. ## TL;DR Checklist - [ ] Select delivery semantics based on idempotency capability of consumers - [ ] Define partitioning strategy that preserves ordering for related messages - [ ] Configure consumer groups with balanced rebalancing strategies - [ ] Implement dead letter queue with poison message detection and retry logic - [ ] Set message TTL to prevent stale messages from accumulating indefinitely - [ ] Match broker choice to workload (Kafka for streaming, RabbitMQ for complex routing, SQS for managed simplicity) --- ## When to Use Use this skill when: - Designing inter-service communication that requires reliable async messaging - Implementing a distributed workflow where message ordering affects correctness - Building event pipelines that must survive broker restarts and network partitions - Configuring consumer groups for horizontal scaling across multiple worker instances - Adding dead letter queue handling to prevent poison messages from blocking queues - Selecting a message broker (Kafka, RabbitMQ, NATS, SQS) for a new system --- ## When NOT to Use Avoid this skill for: - Simple synchronous RPC between services — use HTTP/gRPC instead (see `grpc-patterns`, `rest-api-patterns`) - High-frequency real-time trading where sub-millisecond latency is required — consider in-process event buses - Single-process applications with no distributed components — the broker overhead outweighs benefits - When all consumers are already on a shared memory bus — use an in-process EventBus instead (see `event-bus`) --- ## Core Workflow 1. **Analyze Communication Requirements** — Determine delivery semantics, ordering needs, throughput targets, and retention requirements. **Checkpoint:** If the consumer can safely deduplicate messages, choose at-least-once with idempotency. If zero duplication is required, implement exactly-once with idempotency keys. 2. **Select Queue Topology** — Choose between point-to-point (each message consumed once), pub/sub (each message delivered to all subscribers), fan-out (one producer to many independent queues), or competitive consumers (multiple consumers in one group sharing load). **Checkpoint:** Each consumer group must have a clear purpose — never split a single logical workload across multiple groups. 3. **Define Partitioning Strategy** — Choose partitioning keys that preserve ordering for related messages (e.g., `order_id` for order events, `user_id` for user activity). **Checkpoint:** All consumers in the same group must receive the same set of partitions; rebalancing must not reorder messages within a partition. 4. **Configure Delivery Semantics** — Implement ack-based acknowledgment with configurable timeouts. Set up redelivery policies and dead letter routing. **Checkpoint:** Every message path must have a DLQ — unhandled failures are production incidents, not edge cases. 5. **Implement Consumer Groups** — Define group topology with balanced partition assignment and graceful shutdown handling. **Checkpoint:** Consumers must handle rebalance events by draining in-flight messages before releasing partition ownership. 6. **Set Operational Controls** — Configure TTL for message expiration, priority levels for urgent messages, and delayed delivery for scheduled workflows. **Checkpoint:** All operational controls must have monitoring dashboards — stale queues and blocked DLQs are silent failures. --- ## Implementation Patterns ### Pattern 1: Delivery Semantics — At-Least-Once with Idempotency At-least-once is the safest default. Consumers implement idempotency to safely handle duplicate messages that arrive after a network timeout or broker retry. ```python from __future__ import annotations import hashlib import time import uuid from dataclasses import dataclass, field from typing import Any, Optional @dataclass(frozen=True) class MessageEnvelope: """Immutable message envelope wrapping payload with delivery metadata.""" id: str = field(default_factory=lambda: str(uuid.uuid4())) payload: dict[str, Any] headers: dict[str, str] = field(default_factory=dict) attempt_count: int = 0 delivered_at: float = field(default_factory=time.monotonic) @property def dedup_key(self) -> str: """Deterministic deduplication key combining message id and consumer group.""" return f"{self.id}:{self.headers.get('consumer_group', 'default')}" class IdempotentConsumer: """Processes messages with at-least-once delivery guarantees via deduplication. Uses an in-memory (or backable) store to reject duplicate message IDs within a configurable time window. In production, replace the in-memory store with Redis or a database-backed idempotency table. """ def __init__(self, dedup_ttl_seconds: float = 3600.0) -> None: self._dedup_store: dict[str, float] = {} self._dedup_ttl = dedup_ttl_seconds self._processed_count: int = 0 def process(self, message: MessageEnvelope) -> str: """Process a message, returning result or raising on business failure. Rejects duplicates based on dedup_key. Idempotent — calling this multiple times with the same message returns the cached result. Args: message: The incoming message envelope to process. Returns: A string describing the processing outcome. Raises: RuntimeError: If business logic fails (message will be retried). """ dedup_key = message.dedup_key # Check dedup store — skip already-processed messages if dedup_key in self._dedup_store: return f"DUPLICATE_SKIPPED:{dedup_key}" # Mark as seen self._dedup_store[dedup_key] = time.monotonic() self._cleanup_expired_dedup_entries() try: result = self._execute_business_logic(message) except Exception as exc: # Business failure — message will be redelivered by broker raise RuntimeError(f"Processing failed for {message.id}: {exc}") from exc self._processed_count += 1 return f"PROCESSED:{message.id} -> {result}" def _execute_business_logic(self, message: MessageEnvelope) -> str: """Core business logic — replace with domain-specific implementation.""" action = message.headers.get("action", "unknown") entity_id = message.payload.get("entity_id") return f"action={action} entity={entity_id}" def _cleanup_expired_dedup_entries(self) -> None: """Remove expired deduplication entries to prevent unbounded memory growth.""" now = time.monotonic() expired_keys = [ key for key, seen_at in self._dedup_store.items() if now - seen_at > self._dedup_ttl ] for key in expired_keys: del self._dedup_store[key] class ExactlyOnceProcessor(IdempotentConsumer): """Extends at-least-once with explicit idempotency keys from producers. Producers must include a unique `idempotency_key` header. The consumer deduplicates based on this key rather than the auto-generated message ID, enabling cross-session deduplication (e.g., client retries after timeout). """ def process(self, message: MessageEnvelope) -> str: """Override to use producer-supplied idempotency key.""" idem_key = message.headers.get("idempotency_key") if not idem_key: raise ValueError( "ExactlyOnceProcessor requires 'idempotency_key' header " f"on message {message.id}" ) if idem_key in self._dedup_store: return f"EXACTLY_ONCE_DUPLICATE_SKIPPED:{idem_key}" self._dedup_store[idem_key] = time.monotonic() self._cleanup_expired_dedup_entries() try: result = self._execute_business_logic(message) except Exception as exc: # Do NOT remove from dedup store on failure — retry will re-send same key raise RuntimeError( f"ExactlyOnce processing failed for idempotency_key={idem_key}: {exc}" ) from exc self._processed_count += 1 return f"EXACTLY_ONCE:{message.id} -> {result}" ``` --- ### Pattern 2: Ordered Partitioning by Key Partitioned queues preserve ordering for messages sharing a partition key. All messages with the same key are delivered to the same consumer instance in FIFO order within that partition. ```python from __future__ import annotations import hashlib from dataclasses import dataclass from typing import Protocol @dataclass class OrderedMessage: """A message with an explicit ordering key for partition assignment.""" key: str sequence_number: int # Monotonic within the producer for this key payload: dict class PartitionStrategy(Protocol): """Protocol defining how messages are assigned to partitions.""" def assign_partition(self, key: str, total_partitions: int) -> int: ... class HashPartitionStrategy: """Deterministic partition assignment using consistent hashing. Messages with the same key always map to the same partition. This preserves per-key ordering regardless of consumer rebalancing. """ def __init__(self, num_partitions: int) -> None: if num_partitions < 1: raise ValueError("num_partitions must be >= 1") self._num_partitions = num_partitions def assign_partition(self, key: str, total_partitions: Optional[int] = None) -> int: """Assign a message key to a partition deterministically. Args: key: The partition key (e.g., order_id, user_id). total_partitions: Override default partition count. Returns: An integer partition index in range [0, total_partitions). """ partitions = total_partitions or self._num_partitions hash_value = int(hashlib.sha256(key.encode()).hexdigest(), 16) return hash_value % partitions @property def num_partitions(self) -> int: return self._num_partitions class OrderedPartitionedProducer: """Produces messages with guaranteed per-key ordering via partitioning. Usage: producer = OrderedPartitionedProducer(num_partitions=8) producer.send(OrderedMessage("order-123", 1, {"action": "create"})) producer.send(OrderedMessage("order-123", 2, {"action": "update"})) # Both messages always go to the same partition → ordered delivery. """ def __init__(self, num_partitions: int = 8) -> None: self._strategy = HashPartitionStrategy(num_partitions) @property def num_partitions(self) -> int: return self._strategy.num_partitions def send(self, message: OrderedMessage) -> dict: """Send a message to the appropriate partition. Args: message: The ordered message with key and sequence number. Returns: Metadata about where the message was placed. """ partition = self._strategy.assign_partition(message.key) return { "partition": partition, "key": message.key, "sequence_number": message.sequence_number, "payload": message.payload, } def send_batch( self, messages: list[OrderedMessage], ) -> dict[str, list[dict]]: """Send multiple messages, grouping results by partition. Args: messages: List of ordered messages to batch-send. Returns: Dict mapping partition index to list of placed message metadata. """ by_partition: dict[int, list[dict]] = {} for msg in messages: result = self.send(msg) by_partition.setdefault(result["partition"], []).append(result) return by_partition ``` --- ### Pattern 3: Consumer Groups with Balanced Load Distribution Consumer groups enable horizontal scaling — multiple consumers share partitions and each partition is processed by exactly one consumer in the group at a time. This skill shows balanced assignment with rebalance awareness. ```python from __future__ import annotations import logging import time from collections import defaultdict from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Optional logger = logging.getLogger(__name__) class ConsumerState(Enum): """Lifecycle states for a consumer in a group.""" IDLE = "idle" CONSUMING = "consuming" REBALANCING = "rebalancing" SHUTTING_DOWN = "shutting_down" @dataclass class PartitionAssignment: """Represents which partitions a consumer owns within its group.""" consumer_id: str group_id: str partitions: list[int] = field(default_factory=list) @property def is_empty(self) -> bool: return len(self.partitions) == 0 def add_partitions(self, partitions: list[int]) -> None: for p in sorted(partitions): if p not in self.partitions: self.partitions.append(p) def remove_partitions(self, partitions: list[int]) -> None: self.partitions = [p for p in self.partitions if p not in partitions] class BalancedConsumerGroup: """Manages partition assignment across consumers in a group using round-robin. Handles rebalancing when consumers join or leave the group. During rebalance, all consumers drain in-flight messages before reassignment takes effect. Usage: group = BalancedConsumerGroup("payments", num_partitions=12) consumer = group.register_consumer("worker-1") group.add_consumer(consumer) # ... process messages ... group.remove_consumer(consumer, drain=True) """ def __init__(self, group_id: str, num_partitions: int) -> None: if num_partitions < 1: raise ValueError("num_partitions must be >= 1") self._group_id = group_id self._num_partitions = num_partitions self._consumers: dict[str, BalancedConsumer] = {} self._assignments: dict[str, PartitionAssignment] = {} def register_consumer(self, consumer_id: str) -> BalancedConsumer: """Register a new consumer and return it. Args: consumer_id: Unique identifier for this consumer instance. Returns: A configured Consumer instance ready to be added to the group. """ if consumer_id in self._consumers: raise ValueError(f"Consumer {consumer_id} already registered") consumer = BalancedConsumer(consumer_id, self._group_id) self._consumers[consumer_id] = consumer return consumer def add_consumer(self, consumer: BalancedConsumer) -> dict[int, list[str]]: """Add a consumer to the group and trigger partition rebalancing. Args: consumer: A registered consumer instance. Returns: Mapping of partition index to list of consumer IDs (for verification). """ if consumer.id not in self._consumers: raise ValueError( f"Consumer {consumer.id} not registered with this group" ) consumer.state = ConsumerState.REBALANCING self._rebalance() consumer.state = ConsumerState.CONSUMING return self._get_partition_map() def remove_consumer( self, consumer_id: str, drain: bool = True ) -> dict[int, list[str]]: """Remove a consumer and rebalance partitions among remaining members. Args: consumer_id: The consumer to remove. drain: If True, simulate draining in-flight messages first. Returns: Updated partition map after removal. """ if consumer_id not in self._consumers: raise ValueError(f"Consumer {consumer_id} not found in group") consumer = self._consumers[consumer_id] if drain and not consumer.is_empty: logger.info( "Draining %d in-flight messages for consumer %s before removal", len(consumer.in_flight), consumer_id, ) time.sleep(0.01) # Simulate drain del self._consumers[consumer_id] if consumer.id in self._assignments: del self._assignments[consumer_id] self._rebalance() return self._get_partition_map() def _rebalance(self) -> None: """Reassign all partitions round-robin across active consumers.""" alive_ids = sorted(self._consumers.keys()) if not alive_ids: return # Clear existing assignments for assignment in self._assignments.values(): assignment.partitions.clear() # Round-robin assignment for i, partition_idx in enumerate(range(self._num_partitions)): owner_id = alive_ids[partition_idx % len(alive_ids)] if owner_id not in self._assignments: self._assignments[owner_id] = PartitionAssignment( consumer_id=owner_id, group_id=self._group_id ) self._assignments[owner_id].add_partitions([partition_idx]) def _get_partition_map(self) -> dict[int, list[str]]: """Get a mapping of partition -> owning consumers (for verification).""" partition_map: dict[int, list[str]] = defaultdict(list) for assignment in self._assignments.values(): for p in assignment.partitions: partition_map[p].append(assignment.consumer_id) return dict(partition_map) @property def active_consumers(self) -> int: return len([c for c in self._consumers.values() if c.state == ConsumerState.CONSUMING]) class BalancedConsumer: """A single consumer instance within a group. Tracks its assigned partitions and in-flight messages for rebalance safety. """ def __init__(self, consumer_id: str, group_id: str) -> None: self.id = consumer_id self.group_id = group_id self.state = ConsumerState.IDLE self._assigned_partitions: list[int] = [] self.in_flight: dict[int, Any] = {} @property def is_empty(self) -> bool: return len(self.in_flight) == 0 @property def assigned_partitions(self) -> list[int]: return self._assigned_partitions[:] def assign_partitions(self, partitions: list[int]) -> None: """Update this consumer's partition assignment.""" for p in sorted(partitions): if p not in self._assigned_partitions: self._assigned_partitions.append(p) logger.info( "Consumer %s now owns partitions: %s", self.id, self._assigned_partitions ) def release_partitions(self, partitions: list[int]) -> None: """Release ownership of specified partitions during rebalance. MUST drain in-flight messages before releasing to prevent data loss. """ for p in partitions: if p in self.in_flight: logger.warning( "Releasing partition %d with %d in-flight messages — " "ensure message is requeued", p, len(self.in_flight), ) self._assigned_partitions = [ p for p in self._assigned_partitions if p not in partitions ] ``` --- ### Pattern 4: Dead Letter Queue with Poison Message Handling Dead letter queues (DLQ) catch messages that fail processing after exhausting retries. Poison message detection prevents a single bad message from blocking an entire queue. ```python from __future__ import annotations import logging import time from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional logger = logging.getLogger(__name__) class FailureReason(Enum): """Categorizes why a message processing attempt failed.""" CONSUMER_ERROR = "consumer_error" # Business logic threw SCHEMA_VALIDATION = "schema_validation" # Payload doesn't match expected schema TIMEOUT = "timeout" # Processing exceeded deadline INTEGRATION_FAILURE = "integration_failure" # Downstream service unavailable @dataclass class RetryRecord: """Tracks retry history for a message.""" attempt: int timestamp: float = field(default_factory=time.monotonic) error: Optional[str] = None reason: Optional[FailureReason] = None @property def is_expired(self, max_age_seconds: float = 86400.0) -> bool: """Return True if the retry record has exceeded its time-to-live.""" return (time.monotonic() - self.timestamp) > max_age_seconds @dataclass class DeadLetterMessage: """A message that has exhausted all retries and landed in the DLQ. Contains full metadata for forensic analysis — original payload, headers, retry history, and the final error that caused the move. """ original_message_id: str payload: dict[str, Any] headers: dict[str, str] = field(default_factory=dict) queue_name: str = "" consumer_group: str = "" retry_history: list[RetryRecord] = field(default_factory=list) final_error: Optional[str] = None moved_at: float = field(default_factory=time.monotonic) @property def age_seconds(self) -> float: return time.monotonic() - self.moved_at class PoisonMessageDetector: """Detects and classifies poison messages that repeatedly fail processing. A poison message is one that fails every retry attempt, typically due to malformed data or an irrecoverable downstream dependency failure. This detector classifies failures to enable targeted remediation strategies. """ def __init__(self, max_retries: int = 3) -> None: self._max_retries = max_retries def record_failure( self, message_id: str, error: str, reason: FailureReason, ) -> RetryRecord: """Record a processing failure. Returns the retry record for this attempt.""" return RetryRecord(attempt=1, error=error, reason=reason) # Updated by processor class DeadLetterQueueManager: """Manages dead letter queue operations including poison message detection. After max_retries is exhausted, messages are moved to a DLQ with full metadata for investigation and manual reprocessing. Includes exponential backoff for retry attempts before DLQ routing. """ def __init__( self, max_retries: int = 3, dlq_prefix: str = "dlq_", poison_threshold: int = 5, ) -> None: if max_retries < 1: raise ValueError("max_retries must be >= 1") self._max_retries = max_retries self._dlq_prefix = dlq_prefix self._poison_threshold = poison_threshold self._retry_store: dict[str, list[RetryRecord]] = {} self._failed_consumer_count: dict[str, int] = defaultdict(int) @property def max_retries(self) -> int: return self._max_retries def should_retry(self, message_id: str) -> bool: """Determine if a failed message should be retried. Args: message_id: The ID of the message that just failed. Returns: True if retries remain, False if the message goes to DLQ. """ history = self._retry_store.get(message_id, []) return len(history) < self._max_retries def record_attempt( self, message_id: str, error: str, reason: FailureReason, ) -> RetryRecord | None: """Record a processing attempt and check if DLQ routing is needed. Args: message_id: ID of the message being processed. error: The exception message or error description. reason: Classification of why it failed. Returns: The retry record (always returned), or None if not tracked yet. """ record = RetryRecord( attempt=len(self._retry_store.get(message_id, [])) + 1, error=error, reason=reason, ) self._retry_store.setdefault(message_id, []).append(record) if not self.should_retry(message_id): logger.warning( "Message %s exhausted %d retries → routing to DLQ", message_id, self._max_retries, ) return record # Caller moves this to DLQ return record def get_dlq_name(self, source_queue: str) -> str: """Derive the DLQ name from a source queue name. Args: source_queue: The original queue that messages came from. Returns: DLQ name with prefix (e.g., dlq_orders). """ return f"{self._dlq_prefix}{source_queue}" def get_retry_delay(self, message_id: str) -> float: """Calculate exponential backoff delay for the next retry. Uses base 2^n seconds with a cap of 60 seconds. Returns 0.0 if no retries remain (should go to DLQ instead). Args: message_id: ID of the message being retried. Returns: Delay in seconds before next delivery attempt. """ history = self._retry_store.get(message_id, []) attempt = len(history) + 1 if attempt > self._max_retries: return 0.0 # Exponential backoff: 1s, 2s, 4s, 8s, ... cap at 60s delay = min(2 ** (attempt - 1), 60.0) return delay def get_poison_message_history(self, message_id: str) -> list[RetryRecord] | None: """Get full retry history for a failed message (for DLQ investigation). Args: message_id: ID of the poisoned message. Returns: List of all retry records, or None if message not tracked. """ return self._retry_store.get(message_id) ``` --- ### Pattern 5: Message TTL, Priority Queues, and Delayed Delivery Production systems need to expire stale messages, prioritize urgent work, and schedule delayed delivery for time-based workflows (e.g., "send reminder after 24 hours"). ```python from __future__ import annotations import heapq import logging import time from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) @dataclass(order=True) class PriorityMessage: """A message with an explicit priority level for queue ordering. Lower numeric values indicate higher priority (0 = highest). Uses Python's heapq so the highest-priority message is always first. """ priority: int created_at: float = field(default_factory=time.monotonic, repr=False) message_id: str = field(default="", repr=False) payload: dict[str, Any] = field(default_factory=dict, repr=False) ttl_seconds: Optional[float] = None @property def is_expired(self) -> bool: """Check if this message has exceeded its time-to-live. Returns: True if the message has expired and should be dropped. """ if self.ttl_seconds is None: return False return (time.monotonic() - self.created_at) > self.ttl_seconds class PriorityMessageQueue: """Thread-safe priority queue for messages with TTL support. Messages are dequeued in priority order (lowest number = highest priority). Expired messages are automatically purged during dequeue operations. Usage: queue = PriorityMessageQueue() queue.enqueue(PriorityMessage(priority=1, payload={"urgent": True})) queue.enqueue(PriorityMessage(priority=5, payload={"normal": True})) # dequeue() always returns priority=1 message first """ def __init__(self, max_size: int = 100_000) -> None: self._heap: list[PriorityMessage] = [] self._max_size = max_size self._purged_count: int = 0 self._delivered_count: int = 0 @property def size(self) -> int: return len(self._heap) @property def is_empty(self) -> bool: return len(self._heap) == 0 def enqueue(self, message: PriorityMessage) -> bool: """Add a message to the priority queue. Args: message: The priority message to add. Returns: True if added successfully, False if queue is full. """ if len(self._heap) >= self._max_size: logger.error("Queue full (%d messages), dropping message %s", self._max_size, message.message_id) return False heapq.heappush(self._heap, message) self._delivered_count += 1 return True def dequeue(self) -> PriorityMessage | None: """Remove and return the highest-priority non-expired message. Expired messages are silently purged before returning a result. Returns None if the queue is empty or all messages have expired. Returns: The highest-priority valid message, or None. """ while self._heap: message = heapq.heappop(self._heap) if message.is_expired: self._purged_count += 1 logger.debug( "Purging expired priority message %s (age=%.1fs)", message.message_id, time.monotonic() - message.created_at, ) continue return message return None def peek(self) -> PriorityMessage | None: """Return the highest-priority message without removing it. Returns: The top message or None if queue is empty. """ # Clean expired items at front only (non-destructive for others) while self._heap and self._heap[0].is_expired: heapq.heappop(self._heap) self._purged_count += 1 return self._heap[0] if self._heap else None def purge_expired(self) -> int: """Remove all expired messages from the queue. Returns: The number of messages purged. """ before = len(self._heap) self._heap = [m for m in self._heap if not m.is_expired] heapq.heapify(self._heap) purged = before - len(self._heap) self._purged_count += purged return purged @dataclass class DelayedMessage: """A message scheduled for delivery at a specific future time.""" deliver_at: float # Monotonic timestamp for when this should be delivered message_id: str payload: dict[str, Any] queue_name: str = "" @property def is_ready(self) -> bool: return time.monotonic() >= self.deliver_at class DelayedDeliveryScheduler: """Manages delayed message delivery using a min-heap sorted by deliver_at. Messages are held until their scheduled delivery time, then released for processing. Supports both absolute timestamps and relative delays. Usage: scheduler = DelayedDeliveryScheduler() # Deliver in 5 minutes from now scheduler.schedule(message, delay_seconds=300) # or deliver at specific time scheduler.schedule_at(message, deliver_at=time.monotonic() + 3600) # Periodically check for ready messages ready = scheduler.poll_ready() """ def __init__(self, max_capacity: int = 500_000) -> None: self._schedule: list[DelayedMessage] = [] self._max_capacity = max_capacity @property def pending_count(self) -> int: return len(self._schedule) def schedule(self, message_id: str, payload: dict, delay_seconds: float, queue_name: str = "") -> bool: """Schedule a message for delivery after a relative delay. Args: message_id: Unique identifier for the scheduled message. payload: The message data to deliver. delay_seconds: Seconds from now until delivery is allowed. queue_name: Target queue for the delivered message. Returns: True if scheduled, False if capacity is full. """ if len(self._schedule) >= self._max_capacity: logger.error("Scheduler at capacity (%d), dropping delayed message %s", self._max_capacity, message_id) return False msg = DelayedMessage( deliver_at=time.monotonic() + delay_seconds, message_id=message_id, payload=payload, queue_name=queue_name, ) heapq.heappush(self._schedule, msg) return True def schedule_at(self, message_id: str, payload: dict, deliver_at: float, queue_name: str = "") -> bool: """Schedule a message for delivery at an absolute monotonic timestamp. Args: message_id: Unique identifier for the scheduled message. payload: The message data to deliver. deliver_at: Monotonic timestamp when the message should be delivered. queue_name: Target queue name. Returns: True if scheduled, False if capacity is full. """ if len(self._schedule) >= self._max_capacity: return False msg = DelayedMessage( deliver_at=deliver_at, message_id=message_id, payload=payload, queue_name=queue_name, ) heapq.heappush(self._schedule, msg) return True def poll_ready(self, max_count: int = 100) -> list[DelayedMessage]: """Poll and return all messages that are ready for delivery. Removes delivered messages from the scheduler. Returns at most max_count messages to prevent thundering herds on clock skew. Args: max_count: Maximum messages to return in one poll call. Returns: List of messages whose deliver_at time has passed. """ ready: list[DelayedMessage] = [] remaining: list[DelayedMessage] = [] while self._schedule and len(ready) < max_count: msg = heapq.heappop(self._schedule) if msg.is_ready: ready.append(msg) else: remaining.append(msg) # Put back non-ready messages (they're still pending) for msg in remaining: heapq.heappush(self._schedule, msg) return ready def cancel(self, message_id: str) -> bool: """Cancel a scheduled delayed message by ID. Args: message_id: The ID of the message to cancel. Returns: True if the message was found and removed. """ before = len(self._schedule) self._schedule = [m for m in self._schedule if m.message_id != message_id] heapq.heapify(self._schedule) return len(self._schedule) < before # Was removed if count decreased ``` --- ### Pattern 6: Broker Selection Decision Framework Choosing the right message broker depends on workload characteristics. This framework compares Kafka, RabbitMQ, NATS, and SQS across key dimensions with decision logic for common scenarios. ```python from __future__ import annotations import logging from dataclasses import dataclass, field from enum import Enum from typing import Optional logger = logging.getLogger(__name__) class BrokerCapability(Enum): """Capabilities that distinguish message broker platforms.""" AT_LEAST_ONCE = "at_least_once" EXACTLY_ONCE = "exactly_once" ORDERED_PARTITIONS = "ordered_partitions" PUB_SUB = "pub_sub" POINT_TO_POINT = "point_to_point" COMPLEX_ROUTING = "complex_routing" PRIORITY_QUEUES = "priority_queues" DELAYED_DELIVERY = "delayed_delivery" DEAD_LETTER_QUEUES = "dead_letter_queues" MESSAGE_RETENTION = "message_retention" # Persist and replay history MANAGED_SERVICE = "managed_service" # Fully hosted (AWS SQS, etc.) @dataclass(frozen=True) class BrokerProfile: """Defines the capabilities and trade-offs of a message broker.""" name: str description: str strengths: list[str] weaknesses: list[str] capabilities: list[BrokerCapability] best_for: list[str] recommended_max_throughput_mps: int # messages per second class BrokerSelector: """Decision framework for selecting the right message broker. Evaluates workload requirements against broker profiles to recommend the best-fit platform. Uses weighted scoring for nuanced comparisons. Usage: selector = BrokerSelector() requirements = { "requires_exactly_once": True, "needs_ordered_delivery": True, "managed_service_preferred": False, "max_throughput_mps": 50_000, } recommendation = selector.select(requirements) """ # Broker profiles — the knowledge base for decisions PROFILES: list[BrokerProfile] = [ BrokerProfile( name="Kafka", description="Distributed event streaming platform optimized for high-throughput " "log-based messaging with persistent retention and replay.", strengths=[ "Extremely high throughput (millions of messages/sec)", "Persistent message storage with configurable retention", "Strong ordering guarantees within partitions", "Built-in replication and fault tolerance", "Excellent for event sourcing and data pipelines", ], weaknesses=[ "Complex operational footprint — requires ZooKeeper/KRaft cluster", "Higher latency than pub/sub brokers (~1ms+)", "No native complex routing (direct exchange, headers, etc.)", "Operational overhead for small teams", ], capabilities=[ BrokerCapability.AT_LEAST_ONCE, BrokerCapability.EXACTLY_ONCE, BrokerCapability.ORDERED_PARTITIONS, BrokerCapability.PUB_SUB, BrokerCapability.MESSAGE_RETENTION, ], best_for=[ "Event streaming and data pipelines", "High-throughput event sourcing", "Audit log aggregation", "Real-time analytics feeds", ], recommended_max_throughput_mps=1_000_000, ), BrokerProfile( name="RabbitMQ", description="Flexible message broker with advanced routing capabilities including " "direct exchange, topic exchange, headers, and dead letter queues.", strengths=[ "Rich routing model (direct, topic, headers, fanout exchanges)", "Excellent dead letter queue support", "Priority queue support natively", "Low latency (~1ms for simple delivery)", "Good management UI and monitoring tools", ], weaknesses=[ "Lower throughput than Kafka (tens of thousands msg/sec)", "No persistent message log — messages lost if broker goes down without persistence", "Memory-conscious design requires tuning at scale", "Limited replay capability for consumed messages", ], capabilities=[ BrokerCapability.AT_LEAST_ONCE, BrokerCapability.PUB_SUB, BrokerCapability.POINT_TO_POINT, BrokerCapability.COMPLEX_ROUTING, BrokerCapability.PRIORITY_QUEUES, BrokerCapability.DEAD_LETTER_QUEUES, ], best_for=[ "Complex message routing scenarios", …(truncated)