Skip to content
intermediate Phase 66 · Async Processing

Queue Consumers — Configuration and Management

Configuring and managing queue consumers in Magento 2: consumer configuration, consumer runners, bulk consumption, and failure handling

45m
1 problems
Topic Progress 0%

Consumer Configuration

consumer.xml Structure

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
    <consumer name="product_sync"
              queue="product_sync_queue"
              handler="Vendor\Module\Consumer\ProductSync::process"
              maxMessages="1000"
              maxIdleTime="0"
              sleep="0"
              trxSize="1"
              consumerInstance="Magento\Framework\MessageQueue\ConsumerFactory"
              connection="amqp"/>
</config>

Consumer Options

Option Description
name Consumer identifier
queue Queue to consume from
handler Class::method to process messages
maxMessages Max messages per batch
sleep Seconds between batches
trxSize Messages per transaction
connection amqp or db

Handler Implementation

namespace Vendor\Module\Consumer;

use Psr\Log\LoggerInterface;

class ProductSync
{
    public function __construct(
        private LoggerInterface $logger
    ) {}

    public function process(string $messageBody): void
    {
        $data = json_decode($messageBody, true);
        $this->logger->info('Processing: ' . $data['product_id']);
        $this->syncProduct($data['product_id']);
    }
}

Running Consumers

CLI Commands

# Start a consumer
php bin/magento queue:consumers:start product_sync

# Start with max messages limit
php bin/magento queue:consumers:start product_sync --max-messages=100

# Start with specific connection
php bin/magento queue:consumers:start product_sync --connection=amqp

# List available consumers
php bin/magento queue:consumers:list

Supervisor Configuration

[program:magento-product-sync]
command=php /var/www/html/bin/magento queue:consumers:start product_sync --max-messages=1000
autostart=true
autorestart=true
startsecs=10
startretries=3
user=www-data
stdout_logfile=/var/log/magento/product-sync.log
stderr_logfile=/var/log/magento/product-sync-error.log
numprocs=2

Systemd Service

# /etc/systemd/system/magento-consumer.service
[Unit]
Description=Magento Queue Consumer
After=network.target

[Service]
User=www-data
ExecStart=/usr/bin/php /var/www/html/bin/magento queue:consumers:start product_sync --max-messages=1000
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

Bulk Consumption Patterns

Batch Processing

namespace Vendor\Module\Consumer;

class BatchProcessor
{
    private array $batch = [];
    private int $batchSize = 100;

    public function process(string $messageBody): void
    {
        $this->batch[] = json_decode($messageBody, true);
        
        if (count($this->batch) >= $this->batchSize) {
            $this->processBatch();
            $this->batch = [];
        }
    }

    private function processBatch(): void
    {
        // Bulk insert/update
        $this->connection->insertMultiple('table', $this->batch);
    }
}

Transaction-Based Processing

<consumer name="order_processor"
          queue="order_queue"
          handler="Vendor\Consumer\OrderProcessor::process"
          trxSize="50"/>

Consumer Runner (Cron-Based)

namespace Vendor\Module\Model;

use Magento\MessageQueue\Model\Cron\ConsumerRunner;

class CustomConsumerRunner extends ConsumerRunner
{
    public function run(): void
    {
        // Custom consumer execution logic
        $this->consumer->process(function ($message) {
            $this->handleMessage($message);
        });
    }
}

Scaling Consumers

# Run multiple instances of the same consumer
for i in {1..4}; do
    php bin/magento queue:consumers:start product_sync --max-messages=500 &
done

Failure Handling and Retries

Retry Logic

public function process(string $messageBody): void
{
    $data = json_decode($messageBody, true);
    
    try {
        $this->syncProduct($data['product_id']);
    } catch (TemporaryException $e) {
        // Retry later - re-throw
        throw $e;
    } catch (PermanentException $e) {
        // Log and skip - do not re-throw
        $this->logger->error('Permanent failure: ' . $e->getMessage());
    }
}

Dead Letter Queue

<!-- queue_topology.xml -->
<exchange name="magento-exchange" type="topic">
    <binding id="dlx_binding"
             topic="#"
             destinationType="queue"
             destination="dead_letter_queue"
             arguments>
                <argument name="x-dead-letter-exchange">magento-exchange</argument>
    </binding>
</exchange>

Consumer Health Monitoring

-- Check consumer status
SELECT * FROM queue_consumer
WHERE is_active = 1;

-- Check message backlog
SELECT queue_name, COUNT(*) as pending
FROM queue_message
WHERE status = 'new'
GROUP BY queue_name;

Common Failure Patterns

Pattern Description Fix
Infinite retry Message keeps failing Add max retry count
Poison message Bad data crashes consumer Add validation, DLX
Memory leak Batch grows unbounded Reset batch periodically
Slow consumer Processing takes too long Optimize handler logic
Connection timeout Network issues Add retry with backoff

Practice Problems

0 / 1 solved
Consumer Scaling

A consumer is processing messages too slowly. Design a scaling strategy with multiple instances.

Quiz

1. How do you start a queue consumer?

Question 1 options

2. What does the maxMessages option do?

Question 2 options

3. What happens when a consumer handler throws an exception?

Question 3 options

4. What tool manages consumer processes in production?

Question 4 options

Flashcards

Question

How to start a consumer?

Click to reveal answer

Answer

php bin/magento queue:consumers:start consumer_name

Question

What does maxMessages control?

Click to reveal answer

Answer

Maximum messages processed per batch before consumer stops

Question

What handles failed messages?

Click to reveal answer

Answer

Dead letter queue (DLX) collects messages that exceed retry limits

Question

How to run multiple consumer instances?

Click to reveal answer

Answer

Run the start command multiple times or use supervisor numprocs

Question

What does trxSize control?

Click to reveal answer

Answer

Number of messages processed in a single database transaction

Revision Notes

Key Takeaways

  • 1. consumer.xml defines consumers with handler class, queue, and options
  • 2. supervisor manages consumer processes with auto-restart
  • 3. Batch processing with trxSize improves database performance
  • 4. Dead letter queue prevents poison messages from blocking queues
  • 5. Scale consumers by running multiple instances

Interview Tips

  • • Explain how to configure and run consumers
  • • Discuss failure handling strategies (retry, DLX, skip)
  • • Describe scaling consumers for high throughput
  • • Know the difference between maxMessages and trxSize

Cheat Sheet

Consumers Cheat Sheet

Start:
php bin/magento queue:consumers:start name --max-messages=1000

Supervisor:

[program:magento-consumer]
command=php bin/magento queue:consumers:start name --max-messages=1000
autorestart=true
numprocs=2

Handler:

public function process(string $message): void {
    try { /* work */ }
    catch (TempException $e) { throw $e; }
    catch (PermException $e) { /* log and skip */ }
}

Scale: Run multiple instances or increase numprocs
Monitor: SELECT from queue_message WHERE status = 'new'