Magento 2 Command + Message Queue – asynchroniczne operacje w tle

Magento 2 Command pattern z Message Queue to połączenie dwóch mechanizmów: Command enkapsuluje operację jako obiekt, a Message Queue wykonuje ją asynchronicznie. W praktyce: zamiast wysyłać email synchronicznie w żądaniu HTTP (blokując odpowiedź), wrzucasz wiadomość do kolejki i konsumer obsłuży ją w tle. Pokażę jak to wdrożyć na przykładzie powiadomień o zamówieniu.

Command pattern – podstawy

<?php

declare(strict_types=1);

// Command - enkapsulacja operacji jako obiekt
interface CommandInterface
{
    public function execute(): void;
}

// Konkretna komenda
class SendOrderConfirmationCommand implements CommandInterface
{
    public function __construct(
        private int    $orderId,
        private string $customerEmail,
        private string $customerName,
    ) {}

    public function execute(): void
    {
        // Logika wysyłki emaila
        echo "Wysyłam email do {$this->customerEmail} dla zamówienia #{$this->orderId}\n";
    }

    // Gettery dla serializacji do kolejki
    public function getOrderId(): int        { return $this->orderId; }
    public function getCustomerEmail(): string { return $this->customerEmail; }
    public function getCustomerName(): string  { return $this->customerName; }
}

// Invoker - wie kiedy wykonać komendę, nie wie jak
class CommandBus
{
    private array $handlers = [];

    public function register(string $commandClass, callable $handler): void
    {
        $this->handlers[$commandClass] = $handler;
    }

    public function dispatch(CommandInterface $command): void
    {
        $class = get_class($command);
        if (!isset($this->handlers[$class])) {
            throw new \RuntimeException("No handler for {$class}");
        }
        ($this->handlers[$class])($command);
    }
}

Message Queue w Magento 2 – konfiguracja

<!-- etc/communication.xml - definicja tematu -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework:Communication/etc/communication.xsd">
    <topic name="vendor.module.order.notification"
           is_synchronous="false"
           request="Vendor\Module\Api\Data\OrderNotificationInterface"/>
</config>

<!-- etc/queue_topology.xml - routing wiadomości -->
<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="amqp">
        <binding id="orderNotification"
                 topic="vendor.module.order.notification"
                 destinationType="queue"
                 destination="vendor.module.order.notification"/>
    </exchange>
</config>

<!-- etc/queue_consumer.xml - konfiguracja konsumera -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
    <consumer name="vendor.module.order.notification"
              queue="vendor.module.order.notification"
              connection="amqp"
              maxMessages="100"
              handler="Vendor\Module\Model\Queue\OrderNotificationConsumer::process"/>
</config>

Publisher i Consumer

<?php

declare(strict_types=1);

use Magento\Framework\MessageQueue\PublisherInterface;

// Publisher - wrzuca wiadomość do kolejki
class OrderNotificationPublisher
{
    private const TOPIC = 'vendor.module.order.notification';

    public function __construct(
        private PublisherInterface                          $publisher,
        private \Vendor\Module\Api\Data\OrderNotificationInterfaceFactory $factory,
    ) {}

    public function publish(int $orderId, string $email, string $name): void
    {
        $notification = $this->factory->create();
        $notification->setOrderId($orderId);
        $notification->setCustomerEmail($email);
        $notification->setCustomerName($name);

        // To jest asynchroniczne - metoda zwraca od razu
        $this->publisher->publish(self::TOPIC, $notification);
    }
}

// Consumer - przetwarza wiadomości z kolejki (uruchamiany przez CLI)
class OrderNotificationConsumer
{
    public function __construct(
        private \Vendor\Module\Service\EmailService $emailService,
        private \Psr\Log\LoggerInterface            $logger,
    ) {}

    public function process(\Vendor\Module\Api\Data\OrderNotificationInterface $notification): void
    {
        try {
            $this->emailService->sendOrderConfirmation(
                $notification->getCustomerEmail(),
                $notification->getCustomerName(),
                $notification->getOrderId(),
            );
        } catch (\Exception $e) {
            // Loguj błąd - wiadomość trafi do dead letter queue
            $this->logger->error('Order notification failed', [
                'order_id' => $notification->getOrderId(),
                'error'    => $e->getMessage(),
            ]);
            throw $e; // ponów próbę
        }
    }
}

Uruchomienie konsumera

# Ręczne uruchomienie konsumera
bin/magento queue:consumers:start vendor.module.order.notification

# Z limitem wiadomości (np. dla crona):
bin/magento queue:consumers:start vendor.module.order.notification --max-messages=100

# Lista wszystkich konsumerów
bin/magento queue:consumers:list

# W produkcji - uruchom przez supervisor lub systemd:
# /etc/supervisor/conf.d/magento-consumer.conf:
# [program:magento-consumer]
# command=/usr/bin/php /var/www/html/bin/magento queue:consumers:start vendor.module.order.notification
# autostart=true
# autorestart=true
# numprocs=2  ; dwa procesy równolegle

Podsumowanie

Command + Message Queue to wzorzec dla operacji które: zajmują dużo czasu (email, PDF, integracja z zewnętrznym API), mogą się nie powieść i wymagają ponowienia, lub mogą być wykonane asynchronicznie bez wpływu na UX użytkownika. Magento 2 wspiera zarówno MySQL (dla prostych przypadków bez RabbitMQ) jak i AMQP (RabbitMQ) jako backend kolejki. Dla większości sklepów MySQL queue wystarczy; AMQP potrzebny przy dużym wolumenie wiadomości.