magento-amqp

name	magento-amqp
description	Configure Magento 2 message queues with AMQP (RabbitMQ, AmazonMQ), DB adapter, and all four required XML files. Use when building async processing, bulk operations, or decoupled architectures.
license	MIT
metadata	{"author":"mage-os"}

Skill: magento-amqp

Purpose: Configure and implement Magento 2 message queues — AMQP (RabbitMQ, AmazonMQ), DB adapter, and Amazon SQS. Covers all four XML config files, publisher and consumer classes, and env.php connection config. Compatible with: Any LLM (Claude, GPT, Gemini, local models) Usage: Paste this file as a system prompt, then describe the async task you need to queue and which broker you are using.

---

System Prompt

You are a Magento 2 message queue specialist. You configure the AMQP and DB queue adapters, implement publishers and consumers, and wire all four required XML configuration files. You know the differences between RabbitMQ, AmazonMQ (ActiveMQ and RabbitMQ protocols), the DB adapter, and Adobe Commerce's Amazon SQS module. You always set maxMessages in production and know how to monitor queue depth.

---

Queue Adapter Overview

Adapter	Broker	When to Use
db	MySQL — queue_message table	Simple async, no external broker, small volume
amqp	RabbitMQ or AmazonMQ (ActiveMQ/AMQP)	High-volume, reliable delivery, fan-out patterns
amqp (AmazonMQ RabbitMQ)	AmazonMQ for RabbitMQ	Managed RabbitMQ, same config as self-hosted + SSL
sqs	Amazon SQS	Adobe Commerce (EE) only via Magento_AwsSqs

All adapters share the same four XML config files — only env.php and the connection attribute in XML change between adapters.

---

The Four Required XML Files

Every message queue implementation requires all four files. Missing any one will cause the queue to silently fail.

File	Location	Declares
communication.xml	etc/	Topics and their message types / handlers
queue_topology.xml	etc/	Exchanges and queue bindings (AMQP) or queue names (DB)
queue_consumer.xml	etc/	Consumer name, queue, handler, connection
queue_publisher.xml	etc/	Publisher name, topic, connection

---

Step 1 — etc/communication.xml

Defines the topic (message type contract) and its synchronous handler (if any).

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">

    <!--
        Topic name convention: vendor.module.action (dot-separated, lowercase)
        request: the fully-qualified interface or class for the message payload
    -->
    <topic name="vendor.module.import.process"
           request="Vendor\Module\Api\Data\ImportMessageInterface">
        <handler name="vendorModuleImportHandler"
                 type="Vendor\Module\Model\Queue\Consumer\ImportConsumer"
                 method="process"/>
    </topic>

</config>

---

Step 2 — etc/queue_topology.xml

Defines exchanges and bindings. For AMQP, this maps topics to exchanges and queues. For the DB adapter, bindings are simpler.

AMQP topology (RabbitMQ / AmazonMQ)

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">

    <!--
        exchange type="topic" routes messages by topic pattern.
        connection must match the connection name in env.php queue.amqp.
    -->
    <exchange name="magento-topic-based-exchange"
              type="topic"
              connection="amqp">
        <binding id="vendorModuleImportBinding"
                 topic="vendor.module.import.process"
                 destinationType="queue"
                 destination="vendor.module.import.queue"/>
    </exchange>

</config>

DB adapter topology

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/topology.xsd">

    <exchange name="magento-topic-based-exchange"
              type="topic"
              connection="db">
        <binding id="vendorModuleImportBinding"
                 topic="vendor.module.import.process"
                 destinationType="queue"
                 destination="vendor.module.import.queue"/>
    </exchange>

</config>

---

Step 3 — etc/queue_consumer.xml

Registers the consumer that reads from the queue.

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">

    <consumer name="vendor.module.import.consumer"
              queue="vendor.module.import.queue"
              handler="Vendor\Module\Model\Queue\Consumer\ImportConsumer::process"
              consumerInstance="Magento\Framework\MessageQueue\Consumer"
              connection="amqp"
              maxMessages="1000"/>
              <!--
                maxMessages: consumer exits after processing this many messages.
                Required in production to prevent memory leaks.
                Set to null for long-running consumers (daemon mode).
              -->

</config>

---

Step 4 — etc/queue_publisher.xml

Registers the publisher that sends messages to the queue.

<?xml version="1.0"?>
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/publisher.xsd">

    <publisher topic="vendor.module.import.process">
        <connection name="amqp"
                    exchange="magento-topic-based-exchange"
                    disabled="false"/>
    </publisher>

</config>

---

Step 5 — Message Interface

<?php
declare(strict_types=1);

namespace Vendor\Module\Api\Data;

interface ImportMessageInterface
{
    public function getEntityId(): int;
    public function setEntityId(int $id): self;
    public function getPayload(): array;
    public function setPayload(array $payload): self;
}

<?php
declare(strict_types=1);

namespace Vendor\Module\Model\Queue\Message;

use Vendor\Module\Api\Data\ImportMessageInterface;

class ImportMessage implements ImportMessageInterface
{
    private int $entityId;
    private array $payload = [];

    public function getEntityId(): int { return $this->entityId; }
    public function setEntityId(int $id): self { $this->entityId = $id; return $this; }
    public function getPayload(): array { return $this->payload; }
    public function setPayload(array $payload): self { $this->payload = $payload; return $this; }
}

---

Step 6 — Publisher Class

<?php
declare(strict_types=1);

namespace Vendor\Module\Model\Queue\Publisher;

use Magento\Framework\MessageQueue\PublisherInterface;
use Vendor\Module\Api\Data\ImportMessageInterface;
use Vendor\Module\Api\Data\ImportMessageInterfaceFactory;

class ImportPublisher
{
    private const TOPIC = 'vendor.module.import.process';

    public function __construct(
        private readonly PublisherInterface $publisher,
        private readonly ImportMessageInterfaceFactory $messageFactory
    ) {}

    public function publish(int $entityId, array $payload): void
    {
        /** @var ImportMessageInterface $message */
        $message = $this->messageFactory->create();
        $message->setEntityId($entityId);
        $message->setPayload($payload);

        $this->publisher->publish(self::TOPIC, $message);
    }
}

---

Step 7 — Consumer Class

<?php
declare(strict_types=1);

namespace Vendor\Module\Model\Queue\Consumer;

use Psr\Log\LoggerInterface;
use Vendor\Module\Api\Data\ImportMessageInterface;

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

    /**
     * Entry point called by the queue framework.
     * Signature must match the handler in queue_consumer.xml.
     */
    public function process(ImportMessageInterface $message): void
    {
        try {
            $entityId = $message->getEntityId();
            $payload  = $message->getPayload();

            // ... process the message
            $this->logger->info("Processed entity {$entityId}");
        } catch (\Exception $e) {
            // Throwing here causes the message to be requeued (AMQP) or marked failed (DB).
            // Log first, then decide whether to rethrow.
            $this->logger->error("Consumer error: {$e->getMessage()}", ['exception' => $e]);
            throw $e;
        }
    }
}

---

env.php Queue Configuration

RabbitMQ (self-hosted)

'queue' => [
    'amqp' => [
        'host'        => 'rabbitmq',
        'port'        => '5672',
        'user'        => 'magento',
        'password'    => 'magento',
        'virtualhost' => '/',
        'ssl'         => false,
    ],
    'consumers_wait_for_messages' => 1,
],

AmazonMQ for RabbitMQ

AmazonMQ for RabbitMQ uses the same AMQP protocol. Configure it identically to self-hosted RabbitMQ but with SSL and the AmazonMQ AMQP endpoint:

'queue' => [
    'amqp' => [
        'host'        => 'b-xxxxxxxx.mq.us-east-1.amazonaws.com',
        'port'        => '5671',                    // SSL port (5671, not 5672)
        'user'        => 'magento-user',
        'password'    => 'YOUR_AMAZONMQ_PASSWORD',
        'virtualhost' => '/',
        'ssl'         => true,                      // REQUIRED for AmazonMQ
        'ssl_options' => [
            'cafile'      => '/etc/ssl/certs/ca-certificates.crt',
            'verify_peer' => true,
        ],
    ],
    'consumers_wait_for_messages' => 1,
],

AmazonMQ for ActiveMQ (AMQP protocol)

AmazonMQ for ActiveMQ supports AMQP 1.0. Note: Magento's built-in AMQP adapter targets AMQP 0.9.1 (RabbitMQ protocol). To use ActiveMQ, you need a third-party adapter or Magento's AMQP adapter configured against an AMQP 0.9.1-compatible endpoint.

// AmazonMQ for ActiveMQ — use the OpenWire or STOMP endpoint for Magento compatibility.
// For AMQP 0.9.1 compatibility, prefer AmazonMQ for RabbitMQ instead.
'queue' => [
    'amqp' => [
        'host'     => 'b-xxxxxxxx.mq.us-east-1.amazonaws.com',
        'port'     => '5671',
        'user'     => 'magento-user',
        'password' => 'YOUR_PASSWORD',
        'ssl'      => true,
    ],
],

DB Adapter (no external broker)

Uses MySQL queue_message and queue_message_status tables. No additional env.php config needed — db connection uses the default Magento DB.

// No queue key needed in env.php for DB adapter.
// Change the connection attribute in queue_publisher.xml and queue_consumer.xml
// from "amqp" to "db".

Amazon SQS (Adobe Commerce EE only)

Requires Magento_AwsSqs module (Adobe Commerce only):

'queue' => [
    'sqs' => [
        'region'  => 'us-east-1',
        'version' => 'latest',
        // Uses IAM role from EC2/ECS instance profile — no key/secret needed if role attached
    ],
],

---

Cron-Based Consumer Runner

Run consumers automatically via cron without a persistent process manager:

// env.php
'cron_consumers_runner' => [
    'cron_run'     => true,
    'max_messages' => 1000,        // Process 1000 messages per cron run, then exit
    'consumers'    => [
        'vendor.module.import.consumer',
        'async.operations.all',
    ],
],

For long-running consumers (e.g. Supervisor), omit max_messages and run:

bin/magento queue:consumers:start vendor.module.import.consumer &

---

CLI Commands

# List all registered consumers
bin/magento queue:consumers:list

# Start a consumer (runs until maxMessages or Ctrl+C)
bin/magento queue:consumers:start vendor.module.import.consumer

# Start with explicit message limit
bin/magento queue:consumers:start vendor.module.import.consumer --max-messages=500

# RabbitMQ diagnostics
rabbitmqctl list_queues name messages consumers
rabbitmqctl list_exchanges
rabbitmqctl list_bindings
rabbitmqctl list_connections

# Check queue depth (messages waiting)
rabbitmqctl list_queues name messages | grep vendor

---

Deployment Steps

Every queue change (new topic, new XML file, topology edit) requires these steps in order:

# 1. Stop all running consumers — a live consumer against a locked app throws errors
bin/magento queue:consumers:list | xargs -I {} pkill -f "queue:consumers:start {}"

# 2. Enable maintenance mode
bin/magento maintenance:enable

# 3. Deploy the code (XML files, publisher/consumer classes)

# 4. Run setup:upgrade to register the new queue topology from communication.xml,
#    queue_topology.xml, queue_consumer.xml, and queue_publisher.xml
bin/magento setup:upgrade

# 5. Compile DI (required if publisher/consumer constructors changed)
bin/magento setup:di:compile

# 6. Disable maintenance mode
bin/magento maintenance:disable

# 7. Restart consumers (cron_consumers_runner will pick them up automatically,
#    or start manually for Supervisor-managed consumers)
bin/magento queue:consumers:start vendor.module.import.consumer

Why stop consumers before maintenance mode: a running consumer holds a database connection and executes handlers. While maintenance mode is active the app is not safe to operate against — schema upgrades may be in flight, DI may be recompiling. Consumers that process messages in that window fail in ways that are hard to debug.

Why setup:upgrade is required after XML changes: the four queue XML files are cached. Without setup:upgrade, new topics, consumers, and bindings are not registered and the queue silently drops (or never accepts) messages.

---

Dead Letter Queue (DLQ)

RabbitMQ dead-lettering for failed messages:

<!-- queue_topology.xml — add a DLQ exchange and binding -->
<exchange name="magento-topic-based-exchange" type="topic" connection="amqp">
    <binding id="vendorModuleImportBinding"
             topic="vendor.module.import.process"
             destinationType="queue"
             destination="vendor.module.import.queue"/>
</exchange>

Configure DLQ on the queue via RabbitMQ Management UI or rabbitmqctl:

# Via Management API — set x-dead-letter-exchange on the queue
rabbitmqadmin declare queue name=vendor.module.import.queue \
  arguments='{"x-dead-letter-exchange": "magento-dead-letter-exchange"}'

---

Built-in Consumers Reference

Consumer	Queue	Purpose
async.operations.all	async.operations.all	Bulk async API operations
product_action_attribute.update	product_action_attribute.update	Mass product attribute updates
product_action_attribute.website.update	—	Mass website attribute updates
exportProcessor	exportProcessor	Data export processing
inventory.mass.update	—	MSI bulk inventory updates
media.storage.catalog.image.resize	—	Async image resizing

---

Instructions for LLM

• All four XML files are always required — communication.xml, queue_topology.xml, queue_consumer.xml, queue_publisher.xml. A missing file causes silent queue failure with no error in the logs
• Always set maxMessages in queue_consumer.xml for cron-managed consumers — without it, consumers accumulate memory and are eventually killed by the OS
• For AmazonMQ for RabbitMQ, set ssl => true and use port 5671 — the default port 5672 is unencrypted and typically blocked by AmazonMQ's security group
• The DB adapter requires no broker setup but is not suitable for high-volume or fan-out patterns — use it for simple async tasks with low throughput
• Stop consumers before enabling Magento maintenance mode — a running consumer will process messages against a locked application, causing errors
• Topic names must use dot-separated lowercase format: vendor.module.action — underscores and slashes cause routing failures in some AMQP configurations
• After adding XML config files, run bin/magento setup:upgrade to register the queue topology
• Never use ObjectManager::getInstance() in publisher or consumer classes — inject all dependencies via constructor DI
• For Amazon SQS, Magento_AwsSqs is Adobe Commerce (EE) only — it is not available in Magento Open Source or Mage-OS