Message Queues
Magento's queue framework in four XML files, with RabbitMQ or MySQL behind it and a consumer that does the actual work.
$this->publisher->publish('vendor.feed.sync', $message);A message queue moves work out of the request that asked for it. Code publishes a small message and returns; a separate PHP process picks it up later. Magento uses this for admin exports, asynchronous order placement and the asynchronous REST API.
The words
A topic names a kind of message and fixes its payload type. A publisher sends a message to a topic. An exchange routes it, a queue stores it, and a consumer is a long-running process that reads one queue and calls a handler method for each message.
The four files
| File | Declares |
|---|---|
etc/communication.xml | the topic, its payload type and its handler |
etc/queue_topology.xml | exchanges, and the binding from topic to queue |
etc/queue_consumer.xml | a consumer name, the queue it reads, its handler |
etc/queue_publisher.xml | which connection and exchange a topic publishes to |
Only the publisher file is optional. Without it Magento uses the default connection and the exchange named magento.
RabbitMQ or the database
The default connection is db: messages live in the queue, queue_message and queue_message_status tables. Configuring a broker switches the default to amqp:
bin/magento setup:config:set --amqp-host=localhost --amqp-port=5672 \
--amqp-user=magento --amqp-password=secret --amqp-virtualhost=/The database queue needs nothing extra but polls MySQL; RabbitMQ is the better choice once volume grows.
Making it take effect
Run bin/magento setup:upgrade (it creates the tables and declares exchanges, queues and bindings), then bin/magento cache:flush and bin/magento setup:di:compile if you added classes. Confirm with bin/magento queue:consumers:list.
Example
<!-- app/code/Vendor/Feed/etc/communication.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">
<topic name="vendor.feed.sync" request="Vendor\Feed\Api\Data\SyncItemInterface">
<handler name="syncItem"
type="Vendor\Feed\Model\SyncItemHandler"
method="process"/>
</topic>
</config>When to use it
- Push a price change to an ERP without making the admin save wait for the round trip.
- Fan a nightly catalogue export out across several worker processes.
- Accept a webhook, store the payload, and process it after the response is sent.
- Decide whether a project needs RabbitMQ or can stay on the MySQL queue.
More examples
Routing and the consumer declaration
The binding is the piece people forget: it connects the topic name to a queue name. Set connection="db" to pin this queue to the MySQL tables, or drop the attribute to follow whatever the install is configured for. After bin/magento setup:upgrade and cache:flush, bin/magento queue:consumers:list should print vendor.feed.sync; if it does not, the consumer file was not read.
<!-- app/code/Vendor/Feed/etc/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" type="topic" connection="db">
<binding id="vendorFeedSync"
topic="vendor.feed.sync"
destinationType="queue"
destination="vendor.feed.sync"/>
</exchange>
</config>
<!-- app/code/Vendor/Feed/etc/queue_consumer.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
<consumer name="vendor.feed.sync"
queue="vendor.feed.sync"
handler="Vendor\Feed\Model\SyncItemHandler::process"/>
</config>Publishing a message
PublisherInterface::publish($topicName, $data) is the whole API. The payload must match the request type declared for the topic, or publishing throws InvalidArgumentException; Magento serialises it for you. The data object is created through a generated factory because a data object is not injectable. publish() returns as soon as the message is stored, so never read a result from it.
<?php
// app/code/Vendor/Feed/Model/SyncScheduler.php
declare(strict_types=1);
namespace Vendor\Feed\Model;
use Magento\Framework\MessageQueue\PublisherInterface;
use Vendor\Feed\Api\Data\SyncItemInterfaceFactory;
class SyncScheduler
{
public function __construct(
private readonly PublisherInterface $publisher,
private readonly SyncItemInterfaceFactory $syncItemFactory
) {
}
public function schedule(string $sku, int $storeId): void
{
$item = $this->syncItemFactory->create();
$item->setSku($sku);
$item->setStoreId($storeId);
$this->publisher->publish('vendor.feed.sync', $item);
}
}The handler that receives it
The handler is an ordinary class; the queue decides when it runs. Catch the case where the entity has been deleted since the message was published, because by the time a consumer reads a message the world has moved on. An exception that escapes the handler marks the message failed rather than losing it quietly, so let genuine failures throw and swallow only the cases you have decided to ignore.
<?php
// app/code/Vendor/Feed/Model/SyncItemHandler.php
declare(strict_types=1);
namespace Vendor\Feed\Model;
use Magento\Catalog\Api\ProductRepositoryInterface;
use Magento\Framework\Exception\NoSuchEntityException;
use Psr\Log\LoggerInterface;
use Vendor\Feed\Api\Data\SyncItemInterface;
class SyncItemHandler
{
public function __construct(
private readonly ProductRepositoryInterface $productRepository,
private readonly LoggerInterface $logger
) {
}
public function process(SyncItemInterface $item): void
{
try {
$product = $this->productRepository->get($item->getSku(), false, $item->getStoreId());
} catch (NoSuchEntityException $e) {
$this->logger->warning('feed sync: unknown sku ' . $item->getSku());
return;
}
// hand $product to the remote feed client here
}
}
Discussion