Problem: Out-of-Order Messages

Symptom

Producer sends messages 1, 2, 3 in order. Consumer receives 1, 3, 2. Order violated.

Root Causes

1. Multiple Consumers (Load Balancing)

# Producer sends: 1, 2, 3
[Queue] --> Consumer A (slow)   <-- msg 1 (takes 5s)
        |
        --> Consumer B (fast)   <-- msg 2 (takes 1s) ✓ FINISHES FIRST
        |
        --> Consumer C (medium) <-- msg 3 (takes 2s)

Result: 2, 3, 1 (order violated)

2. Message Redelivery After Failure

Timeline:
1. Msg 1 processed ✓
2. Msg 2 delivered, consumer CRASHES before ack
3. Msg 3 processed ✓
4. Msg 2 redelivered (requeued)
5. Msg 2 processed ✓

Final order: 1, 3, 2

3. Parallel Processing Within Consumer

# ❌ BAD - Threads complete out-of-order
def callback(ch, method, properties, body):
    thread = threading.Thread(target=process, args=(body,))
    thread.start()  # Non-blocking, order not preserved

Solutions

✅ Solution 1: Single Consumer

Only ONE consumer processes queue. Guarantees FIFO order.

# Only run ONE instance of this consumer
channel.basic_consume(
    queue='strict_order_queue',
    on_message_callback=callback,
    auto_ack=False
)

def callback(ch, method, properties, body):
    # Process synchronously (blocking)
    result = process_message(body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.start_consuming()  # Run only ONE instance

✅ Solution 2: Sequence Numbers + Reordering Buffer

Producer adds sequence number to each message. Consumer buffers out-of-order messages and reorders them.

# Producer
sequence = 0
for msg in messages:
    sequence += 1
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=json.dumps({'seq': sequence, 'data': msg}).encode()
    )

# Consumer
expected_seq = 1
buffer = {}  # {seq: message}

def callback(ch, method, properties, body):
    msg = json.loads(body.decode())
    seq = msg['seq']

    # Buffer message
    buffer[seq] = msg

    # Process all consecutive messages
    global expected_seq
    while expected_seq in buffer:
        process_message(buffer[expected_seq])
        del buffer[expected_seq]
        expected_seq += 1

    ch.basic_ack(delivery_tag=method.delivery_tag)

✅ Solution 3: Separate Queues Per Entity

Use routing key to send related messages to same queue. Different entities can process in parallel, but same entity always ordered.

# Producer
channel.basic_publish(
    exchange='direct_exchange',
    routing_key=f'user_{user_id}',  # Route by user ID
    body=message.encode()
)

# Consumer binds to specific user queue
channel.queue_bind(
    exchange='direct_exchange',
    queue='user_123_queue',
    routing_key='user_123'
)

# Result: All messages for user 123 processed in order
# Messages for different users can process in parallel

✅ Solution 4: prefetch_count=1 + Synchronous Processing

Limit consumer to ONE message at a time. Process synchronously (no threads).

channel.basic_qos(prefetch_count=1)  # Only 1 message at a time

def callback(ch, method, properties, body):
    # Synchronous processing (blocking)
    result = process_message(body)
    ch.basic_ack(delivery_tag=method.delivery_tag)
    # Next message only delivered AFTER ack

channel.basic_consume(
    queue='ordered_queue',
    on_message_callback=callback,
    auto_ack=False
)

When Ordering Matters vs. Doesn't

🔴 Strict Ordering Required

• Bank transactions (withdraw MUST happen after deposit)
• State machine updates (status: pending → approved → shipped)
• Database migrations (schema changes must apply in order)
• Event sourcing (events must replay in order)

🟢 Ordering NOT Required

• Image resizing (each image independent)
• Email sending (order doesn't matter)
• Log aggregation (timestamps handle ordering)
• Independent API calls

Testing Strategy

1. Inject Artificial Delays:

   # Make early messages slow
   if seq == 1:
       time.sleep(5)  # Delay first message

2. Start Multiple Consumers: Send 100 messages with multiple workers. Check if order violated.
3. Kill Consumer Mid-Processing: Force redelivery and check ordering.

Summary