Messaging Patterns

Why Messaging Patterns Matter

Every messaging system on this planet is some combination of three primitives: a producer writes, a broker stores, a consumer reads. The interesting part is what each of those steps does on the unhappy path — the producer’s ack, the broker’s replication, the consumer’s commit. Patterns are how we name the combinations.

This tutorial is about those patterns themselves — the trade-offs, the failure modes, the configuration knobs. It is not about which broker to pick (that’s in the comparison table) and it is not about event-driven architecture as a whole (that’s the next tutorial). The lens here is narrow on purpose: given that you’re using a broker, what does correct use look like?

Point-to-Point vs Publish-Subscribe

The two shapes look similar in code — produce a message, consume a message — but the broker is doing something fundamentally different in each.

When to use which

• Use a queue when the work has one owner. “Resize this image,” “send this email,” “process this payment.” You want competing consumers, not duplicated work.
• Use a topic when the event has many interested parties — or might in the future. “OrderPlaced” gets handled by billing, by inventory, by analytics, by the customer-comms service. Adding a fifth subscriber should not require touching the producer.
• Don’t use pub/sub for work distribution. If two services both consume the same topic and both try to send the welcome email, the user gets two welcome emails. The fix is one consumer group, not two services subscribing.

Request-Reply over Messaging

Request-reply over messaging is RPC dressed up in async clothing. It has its place — but it’s also where most messaging-as-RPC anti-patterns live.

# Python + pika (RabbitMQ): request-reply with a temporary reply queue
import uuid, time, pika, json

connection = pika.BlockingConnection(pika.ConnectionParameters("broker"))
channel = connection.channel()

# Exclusive, auto-delete reply queue scoped to this client
result = channel.queue_declare(queue="", exclusive=True)
reply_queue = result.method.queue

responses = {}

def on_reply(ch, method, props, body):
    responses[props.correlation_id] = json.loads(body)

channel.basic_consume(queue=reply_queue, on_message_callback=on_reply, auto_ack=True)

def call(payload, timeout=5.0):
    correlation_id = str(uuid.uuid4())
    channel.basic_publish(
        exchange="",
        routing_key="price.quote.requests",
        properties=pika.BasicProperties(
            reply_to=reply_queue,
            correlation_id=correlation_id,
            expiration=str(int(timeout * 1000)),  # server-side TTL on the request
        ),
        body=json.dumps(payload),
    )
    deadline = time.monotonic() + timeout
    while correlation_id not in responses and time.monotonic() < deadline:
        connection.process_data_events(time_limit=0.1)
    return responses.pop(correlation_id, None)

Request-reply does shine when you genuinely benefit from buffering: long-running enrichment jobs, ML inference where queue depth signals when to autoscale workers, batch RPC against a slow legacy system that you don’t want to drown.

Delivery Guarantees

Idempotent Kafka consumer in Python

from confluent_kafka import Consumer, KafkaError
import json, psycopg2

consumer = Consumer({
    "bootstrap.servers": "broker:9092",
    "group.id": "order-processor",
    "enable.auto.commit": False,           # commit ourselves, after the work
    "auto.offset.reset": "earliest",
    "isolation.level": "read_committed",
})
consumer.subscribe(["orders.created"])

db = psycopg2.connect("dbname=orders")

def handle(msg):
    event = json.loads(msg.value())
    idem_key = event["order_id"]            # natural idempotency key
    with db.cursor() as cur:
        # INSERT ... ON CONFLICT DO NOTHING is the heart of idempotency.
        # If we’ve seen this order_id before, the row already exists,
        # rowcount is 0, and we skip the side effects.
        cur.execute(
            """
            INSERT INTO processed_orders (order_id, processed_at)
            VALUES (%s, NOW()) ON CONFLICT (order_id) DO NOTHING
            """,
            (idem_key,),
        )
        if cur.rowcount == 0:
            return                          # duplicate; safe to skip
        charge_card(event)
        send_confirmation(event)
    db.commit()

try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None: continue
        if msg.error():
            if msg.error().code() != KafkaError._PARTITION_EOF:
                raise Exception(msg.error())
            continue
        handle(msg)
        consumer.commit(msg, asynchronous=False)   # commit AFTER success
finally:
    consumer.close()

Two lines do all the heavy lifting: enable.auto.commit=false and the explicit commit(msg) after processing. With auto-commit on, the offset advances on a timer and a crash mid-handler leaves you with a permanently lost message. With manual commit, a crash means the next consumer re-reads the same message — which is fine, because the idempotency key short-circuits the second attempt.

Ordering

Brokers that scale (Apache Kafka, Apache Pulsar, Kinesis, GCP Pub/Sub with ordering keys) all use the same trick: they shard the topic into partitions and guarantee order within a partition, not across. The partition is selected by hashing a key. Pick the key carefully and you get the order you need without giving up parallelism.

Dead-Letter Queues and Poison Messages

DLQs are not optional. The first time you skip one, a malformed message takes down a partition for the weekend.

RabbitMQ: dead-letter exchange

# RabbitMQ definitions: a primary queue with a DLX wired in
exchanges:
  - name: orders
    type: direct
    durable: true
  - name: orders.dlx                   # dead-letter exchange
    type: direct
    durable: true

queues:
  - name: orders.process
    durable: true
    arguments:
      x-dead-letter-exchange: "orders.dlx"
      x-dead-letter-routing-key: "failed"
      x-message-ttl: 600000           # 10 min before DLQ on TTL expiry
      x-delivery-limit: 5             # quorum-queue retry budget
  - name: orders.dlq
    durable: true

bindings:
  - source: orders
    destination: orders.process
    routing_key: "new"
  - source: orders.dlx
    destination: orders.dlq
    routing_key: "failed"

The consumer rejects with requeue=false after exhausting retries. The broker auto-routes the message to orders.dlx with the original payload plus an x-death header explaining why it died. That header is gold for debugging — it carries the original queue, the rejection reason, the count.

Kafka: retry topics, not in-place retries

Apache Kafka has no built-in DLQ. The community pattern, popularized by Uber and Confluent, is a chain of retry topics:

# Topic chain for at-least-once with bounded retries
orders.created                # primary; consumer reads here
orders.created.retry.5s       # 5s delay, then re-attempt
orders.created.retry.30s      # 30s delay
orders.created.retry.5m       # 5m delay
orders.created.dlq            # terminal; alarm on depth

# Pseudo-code in the consumer:
def handle(msg):
    try:
        process(msg)
    except RetryableError:
        attempt = int(msg.headers().get("retry-attempt", 0))
        if attempt >= 3:
            producer.produce("orders.created.dlq", msg.value(),
                             headers={"failure-reason": "max-retries"})
        else:
            next_topic = ["retry.5s", "retry.30s", "retry.5m"][attempt]
            producer.produce(f"orders.created.{next_topic}", msg.value(),
                             headers={"retry-attempt": str(attempt + 1)})
    except NonRetryableError as e:
        producer.produce("orders.created.dlq", msg.value(),
                         headers={"failure-reason": str(e)})
    consumer.commit(msg)            # always commit; the message moved on

AWS SQS: a DLQ is a redrive policy

# Terraform: AWS SQS queue with redrive to DLQ after 5 failed receives
resource "aws_sqs_queue" "orders_dlq" {
  name                       = "orders-dlq"
  message_retention_seconds  = 1209600      # 14 days; max
}

resource "aws_sqs_queue" "orders" {
  name                        = "orders"
  visibility_timeout_seconds  = 60           # must exceed handler P99
  message_retention_seconds   = 345600       # 4 days
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.orders_dlq.arn
    maxReceiveCount     = 5                  # 5 deliveries, then DLQ
  })
}

# CloudWatch alarm on DLQ depth — the only signal that matters
resource "aws_cloudwatch_metric_alarm" "dlq_depth" {
  alarm_name          = "orders-dlq-not-empty"
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  statistic           = "Maximum"
  period              = 60
  evaluation_periods  = 1
  threshold           = 0
  comparison_operator = "GreaterThanThreshold"
  dimensions = { QueueName = aws_sqs_queue.orders_dlq.name }
}

Backpressure and Flow Control

Different brokers expose flow control differently, but the pattern is the same: bound the in-flight work per consumer, watch the lag, scale.

NATS JetStream: a durable pull consumer with bounded in-flight

// Go: a durable JetStream consumer with strict flow control
package main

import (
    "context"
    "time"
    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

func main() {
    nc, _ := nats.Connect("nats://broker:4222")
    js, _ := jetstream.New(nc)
    ctx := context.Background()

    // Stream backs the topic; defines retention and replication
    js.CreateStream(ctx, jetstream.StreamConfig{
        Name:      "ORDERS",
        Subjects:  []string{"orders.>"},
        Replicas:  3,
        Storage:   jetstream.FileStorage,
        Retention: jetstream.WorkQueuePolicy,        // queue semantics
        MaxAge:    72 * time.Hour,
    })

    cons, _ := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
        Durable:        "order-processor",
        AckPolicy:      jetstream.AckExplicitPolicy,
        MaxAckPending:  100,                          // hard ceiling on in-flight
        AckWait:        30 * time.Second,
        MaxDeliver:     5,                            // then -> DLQ stream
        BackOff:        []time.Duration{1 * time.Second, 5 * time.Second, 30 * time.Second},
    })

    // Pull batches; never accept more than we can chew
    iter, _ := cons.Messages(jetstream.PullMaxMessages(25))
    for {
        msg, err := iter.Next()
        if err != nil { break }
        if err := handle(msg.Data()); err == nil {
            msg.Ack()
        } else {
            msg.NakWithDelay(5 * time.Second)   // negative ack with backoff
        }
    }
}

Broker Comparison

Every broker is a different set of trade-offs. The table below is opinionated — defaults and typical operational profiles, not the absolute upper bounds you can tune to.

Best Practices