Message Queues Deep Dive

Delivery Modes

Mode              | Guarantee           | Use Case
──────────────────|─────────────────────|──────────────────
At-most-once      | Message may be lost | Metrics, logging
At-least-once     | No loss, may dup    | Orders, payments
Exactly-once      | No loss, no dup     | Critical operations

At-Least-Once Implementation

// Publish with persistence
$channel->basic_publish($message, $exchange, false, true);
// persistent = true, mandatory = false

// Acknowledge after processing
$channel->basic_consume($queue, '', false, false, false, false,
    function ($msg) {
        try {
            $this->process($msg->body);
            $msg->delivery_info['channel']->basic_ack(
                $msg->delivery_info['delivery_tag']
            );
        } catch (Exception $e) {
            // Requeue on failure
            $msg->delivery_info['channel']->basic_nack(
                $msg->delivery_info['delivery_tag'],
                false,
                true  // requeue
            );
        }
    }
);

Exactly-Once Patterns

1. Idempotent processing
2. Transactional outbox
3. Saga pattern with compensation
4. Two-phase commit

Ordering Challenges

Problem: Messages processed out of order
1. Order placed: $100
2. Order updated: $150  
3. Order processed: $100 (stale!)

Ordering Strategies

Strategy           | How                     | Trade-off
───────────────────|─────────────────────────|─────────────
Partition by key    | Same key → same queue   | Limited parallelism
Global ordering     | Single queue            | No parallelism
Sequence numbers   | Consumer sorts          | Complex consumer

Partitioned Ordering

// Route by entity ID for ordering
$orderId = $message->getOrderId();
$partitionId = $orderId % $totalPartitions;

$channel->basic_publish(
    $message,
    $exchange,
    false,
    false,
    [
        'routing_key' => 'orders.partition.' . $partitionId
    ]
);

// Each partition has its own consumer
// Same order always goes to same partition

DLQ Architecture

Main Queue → Consumer → Process
    │                    │
    │ Success: ACK       │ Failure: NACK
    │                    │
    â–¼                    â–¼
  Delete            Dead Letter Queue
                         │
                         â–¼
                    Manual Review
                    or Retry

DLQ Configuration

// RabbitMQ DLQ
$channel->queue_declare(
    'orders.main',
    false, true, false, false,
    false,
    [
        'x-dead-letter-exchange' => 'dlx',
        'x-dead-letter-routing-key' => 'orders.dlq',
        'x-message-ttl' => 300000  // 5 min TTL
    ]
);

// DLQ consumer
$channel->basic_consume('orders.dlq', '', false, false, false, false,
    function ($msg) {
        // Log for investigation
        $this->logger->error('DLQ message', [
            'queue' => $msg->delivery_info['routing_key'],
            'body' => $msg->body,
            'retries' => $msg->delivery_info['redelivered']
        ]);
        
        // Store for manual review
        $this->dlqService->store($msg);
        $msg->delivery_info['channel']->basic_ack(
            $msg->delivery_info['delivery_tag']
        );
    }
);

DLQ Handling Strategies

1. Automatic retry with backoff
2. Manual review queue
3. Alert after threshold
4. Archive after max retries
5. Delete after review

Reliability Patterns

Pattern              | Purpose
─────────────────────|──────────────────────
Idempotent consumer  | Handle duplicate messages
Transaction outbox   | Atomic DB + message
Saga pattern         | Distributed transactions
Circuit breaker      | Prevent cascade failures

Transaction Outbox

// Publish message atomically with DB
function placeOrder($orderData) {
    $this->db->beginTransaction();
    
    try {
        // 1. Save order to DB
        $order = $this->orderService->create($orderData);
        
        // 2. Save message to outbox table
        $this->outbox->save([
            'id' => uniqid(),
            'topic' => 'order.placed',
            'payload' => json_encode($order->toArray()),
            'status' => 'pending'
        ]);
        
        $this->db->commit();
    } catch (Exception $e) {
        $this->db->rollBack();
        throw $e;
    }
}

// Background worker publishes from outbox
function publishPendingMessages() {
    $messages = $this->outbox->findByStatus('pending');
    foreach ($messages as $msg) {
        $this->publisher->publish($msg->topic, $msg->payload);
        $this->outbox->update($msg->id, ['status' => 'published']);
    }
}