Message Queues — Asynchronous Processing in Magento 2

How Message Queues Work

Publisher -> Message -> Queue -> Consumer -> Handler

1. Publisher sends a message to a queue
2. Message is stored in the queue (DB or AMQP)
3. Consumer picks up the message
4. Handler processes the message

Magento Queue Implementation

Magento uses the Magento\Framework\MessageQueue framework:

• Publisher: sends messages
• Queue: stores messages
• Consumer: receives messages
• Handler: processes messages

Queue Types

Basic Configuration

<!-- queue_topology.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
    <exchange name="magento-exchange" type="topic">
        <binding id="vendor_product_sync"
                 topic="vendor.product.sync"
                 destinationType="queue"
                 destination="vendor_product_sync_queue"/>
    </exchange>
</config>

queue_topology.xml

Defines exchanges and bindings:

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">
    <exchange name="magento-exchange" type="topic">
        <binding id="sync_binding"
                 topic="vendor.product.sync"
                 destinationType="queue"
                 destination="product_sync_queue"/>
    </exchange>
</config>

queue_publisher.xml

Configures message publishers:

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/publisher.xsd">
    <publisher id="db" queue="product_sync_queue"/>
</config>

queue_consumer.xml

Configures message consumers:

<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
    <consumer name="product_sync_consumer"
              queue="product_sync_queue"
              handler="Vendor\Module\Model\MessageHandler::process"
              maxMessages="1000"
              maxIdleTime="0"
              sleep="0"/>
</config>

queue_consumer.xml Options

Basic Publisher

namespace Vendor\Module\Service;

use Magento\Framework\MessageQueue\PublisherInterface;

class ProductPublisher
{
    public function __construct(
        private PublisherInterface $publisher
    ) {}

    public function publish(int $productId): void
    {
        $this->publisher->publish(
            'vendor.product.sync',
            json_encode(['product_id' => $productId])
        );
    }
}

Topic-Based Publishing

// Using topic name directly
$this->publisher->publish(
    'vendor.product.sync',
    $messageBody // JSON string
);

// Topic maps to exchange via queue_topology.xml

Batch Publishing

public function publishBatch(array $productIds): void
{
    foreach ($productIds as $id) {
        $this->publisher->publish(
            'vendor.product.sync',
            json_encode(['product_id' => $id])
        );
    }
}

Publishing from Observer

namespace Vendor\Module\Observer;

class ProductSaveObserver implements ObserverInterface
{
    public function __construct(
        private PublisherInterface $publisher
    ) {}

    public function execute(Observer $observer)
    {
        $product = $observer->getEvent()->getProduct();
        $this->publisher->publish(
            'vendor.product.sync',
            json_encode(['product_id' => $product->getId()])
        );
    }
}

Consumer Handler

namespace Vendor\Module\Model;

use Psr\Log\LoggerInterface;

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

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

Running Consumers

# Run a specific consumer
php bin/magento queue:consumers:start product_sync_consumer

# Run with max messages
php bin/magento queue:consumers:start product_sync_consumer --max-messages=100

# Run all consumers
php bin/magento queue:consumers:start

Consumer in Production

# Supervisor configuration
[program:magento-product-sync]
command=php /var/www/html/bin/magento queue:consumers:start product_sync_consumer --max-messages=1000
autostart=true
autorestart=true

Failure Handling

public function process(string $messageBody): void
{
    try {
        $data = json_decode($messageBody, true);
        $this->syncProduct($data['product_id']);
    } catch (\Exception $e) {
        $this->logger->error('Message processing failed: ' . $e->getMessage());
        throw $e; // Re-throw to retry
    }
}