Queue Scaling

Consumer Scaling Architecture

┌─────────────────────────────────────────┐
│            Message Queue (RabbitMQ)      │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐  │
│  │ Queue A │ │ Queue B │ │ Queue C │  │
│  └────┬────┘ └────┬────┘ └────┬────┘  │
└───────┼───────────┼───────────┼────────┘
        │           │           │
   ┌────┴────┐ ┌────┴────┐ ┌────┴────┐
   │Consumer1│ │Consumer2│ │Consumer3│
   │Consumer2│ │Consumer3│ │Consumer4│
   └─────────┘ └─────────┘ └─────────┘

Magento Queue Consumer Config

// app/etc/env.php
'queue' => [
    'consumers' => [
        'product.action.update' => [
            'maxMessages' => 1000,
            'consumer' => 'product.action.update'
        ],
        'codegeneratorProcessor' => [
            'maxMessages' => 100
        ]
    ]
],

Running Multiple Consumers

# Start multiple consumer processes
bin/magento queue:consumers:start product.action.update --max-messages=1000 &
bin/magento queue:consumers:start product.action.update --max-messages=1000 &
bin/magento queue:consumers:start product.action.update --max-messages=1000 &

# Or use supervisor
# /etc/supervisor/conf.d/magento-consumer.conf
[program:magento-consumer-product]
command=bin/magento queue:consumers:start product.action.update --max-messages=1000
process_name=%(program_name)s_%(process_num)02d
numprocs=3
autostart=true
autorestart=true

Partitioning Strategies

Strategy        | Description              | Use Case
────────────────|──────────────────────────|──────────────
Round Robin     | Messages rotated         | Equal processing
Hash-based      | Hash(key) % partitions   | Ordered processing
Priority        | Separate priority queues  | Critical messages
Topic           | Separate topics per type | Different workloads

Topic-Based Partitioning

Topic: catalog.update
├── Partition 0: Product updates
├── Partition 1: Category updates
└── Partition 2: Attribute updates

Each partition consumed independently
Enables parallel processing per type

Hash-Based Partitioning

// Partition by product ID for ordered processing
$productId = $message->getProductId();
$partitionId = $productId % $totalPartitions;

// Route to specific queue
$queueName = 'catalog.update.partition.' . $partitionId;
$publisher->publish($queueName, $message);

Throughput Configuration

# RabbitMQ tuning
# Increase prefetch count for batch processing
rabbitmqctl set_policy prefetch-count 50 "^magento" '{"prefetch-count": 50}'

# Increase memory limit
rabbitmqctl set_vm_memory_high_watermark 0.6

# Disk free limit
rabbitmqctl set_disk_free_limit 2GB

Batch Processing

// Process messages in batches
$batchSize = 50;
$messages = [];

while ($message = $queue->dequeue()) {
    $messages[] = $message;
    
    if (count($messages) >= $batchSize) {
        $this->processBatch($messages);
        $messages = [];
    }
}

// Process remaining
if (!empty($messages)) {
    $this->processBatch($messages);
}

Performance Metrics

Metric                | Target
──────────────────────|────────────
Messages/sec          | >1000
Queue depth           | <10000
Consumer lag          | <100 messages
Processing latency    | <100ms
Ack time              | <50ms

Backpressure Strategies

1. Queue Depth Limit: Reject when queue full
2. Rate Limiting: Throttle message intake
3. Consumer Scaling: Auto-scale consumers
4. Circuit Breaker: Stop intake on failure

Queue Depth Alerts

# Alert if queue depth > 10000
rabbitmqctl list_queues name messages | while read queue msgs; do
    if [ $msgs -gt 10000 ]; then
        echo "ALERT: Queue $queue depth $msgs"
    fi
done

Auto-Scaling Consumers

#!/bin/bash
# Auto-scale based on queue depth
DEPTH=$(rabbitmqctl list_queues -q magento_catalog | awk '{print $2}')
CURRENT=$(supervisorctl status | grep -c "RUNNING")

if [ $DEPTH -gt 5000 ] && [ $CURRENT -lt 10 ]; then
    supervisorctl start magento-consumer
elif [ $DEPTH -lt 1000 ] && [ $CURRENT -gt 2 ]; then
    supervisorctl stop magento-consumer
fi

Dead Letter Queue Handling

Message fails 3 times → Dead Letter Queue

Monitor DLQ depth
Alert if DLQ growing
Reprocess or discard DLQ messages