How to Use RabbitMQ Queue in Extensions For Magento 2.0 EE

Quite often we may need to run in background some resource demanding task to avoid interfering with the main task that is being performed at the same time. It is quite easy to arrange this with the help of queues: we put a task in a queue, and the system carries it out in accordance with the amount of available resources. It is possible to arrange such queues using third party applications, such as RabbitMQ or Gearman. You can also do the job perfectly well even without such third party tools. For instance, instead of RabbitMq for Magento 2.0 EE perfectly suits Mysql + cron, but this solution may have certain disadvantages if compared to real queue-solutions: first off, cron is running as a cycle so you are not able to immediately react on adding of a record into the queue. Secondly, using such a scheme makes it quite difficult to parallel a queue. Magento 2.0 EE gives you the possibility to use queues not only on the basis of Mysql+cron, but also on the basis of RabbitMq.

RabbitMQ allows applications to interact through the AMQP protocol. So to use RabbitMQ we will need to install the library that supports this protocol. Magento 2.0 EE uses the – php-amqplib library that is provided by one of the RabbitMq developers. This library uses AMQ Protocol version 0-9-1 and requires the installation of the extension bcmath in php. If your Magento 2.0 EE is properly installed, then you should have this extension already available (the docker file php7 for Magento 2.0 can be downloaded from https://github.com/Galillei/php7-magento2ee/blob/master/Dockerfile).

For more detailed info about RabbitMq please visit the official website.

But in this article we will describe only those features that can be used in Magento 2 EE.

Magento 2.0 EE uses the pattern Publish/Subscribe (publish/subscribe). I.e. there is a method that is adding records to the Publisher queue and there is a scriptwhich is reading data from the Consumer queue.  Besides, it is even possible to use several Consumer subscribers when each of them will have its own queue, where messages from Publisher will be delivered.

The very first thing we need to do to create a queue is to add the file queue.xml  into the module folder <magento2ee_dir>/app/code/<module_name>/etc/queue.xml:

So, lets check every string of this file.

This line indicates that we are using the default Publisher bridge.set.orders, where the param “schema” stands for interface. This interface defines the type of the message. The object implements this interface and should be delivered to the Publisher, while Consumer should also receive exactly the same object.

In <magento2ee_dir>/app/code/<module_name>/etc/di.xml we have to indicate the class which will implement this interface, so that Magento would know which class should be initialized.  <topic> indicates the type of exchanges (points of exchange) which is being used.

When we add a message to Publisher it does not know anything about queues or subscribed Consumers. In other words, for Publisher it does not matter what or how many Consumers are subscribed for receiving messages. So, this is exactly what exchanges do: distributing messages among Consumers.

There can be different types of Exchange:  topic, direct, fanout and others. The most advanced type is “topic”: it has all features that other types of exchanges have. For instance, this type can use a combined routing for queues (something similar to аn URL) which gives you access to messages. For more details read this article.

In case you need to process and array instead of an object, then instead of:

use, for example:

The line:

Here there are declared Consumer, queue name, queue adapter (amqp). To use mysql instead of RabbitMq you need just to change the parameter connection from amqp to db and to change the class which object will process messages in queue in the method processMessage. The parameter executor indicates the class, which will parse our messages and deliver them to the class MyModule\Consumer\ConsumerOrder

The line:

attaches the queue bridge_sender_queue to exchange=”magento”.

So now if we run the following command in console:

we should get a list with the name of our Consumer, specifically bridgeSender.

To set Consumer listen to the queue, we have to run the following command:

But if try to run the command you may get an error. The thing is that Magento does not create a new queue when either a Consumer is connected or when a message is added to the queue. So if you try to send a message to a non-existing queue nothing will happen and the message will be ignored. And if you try to run the command

then you will receive an error message that such queue does not exist . So, to be able to use the queue we need to bind it to our exchange. Magento uses only topic exchange type, which has the name ‘magento’. To create a queue, you may use the following code in the file <magento2ee_dir>/app/code/<module_name><Setup>/InstallData.php (basically, you may use it in any file from the Setup folder, but I think it is quite logical to use it exactly in the file InstallData.php):

And you should not worry that it will re-create already existing queues. Instead, it will create only those queues that have not existed yet.

So, if you try to run the command now:

Partner With Us Let's discuss how to grow your business. Get a Free Quote.

Talk to Igor

you will get the error – “Not found merger for consumer name  bridgeSender'” . Because for each queue we need to declare the class, that has been inherited from the interface Magento\Framework\MessageQueue\MergerInterface. This class can be used for preliminary processing of messages even before the class that was declared in queue.xml in the Consumer section. It is obligatory to use this class, but if you do not require preliminary processing of messages, then you may simply return messages back:

Having created this file, try to run the command once again:

And again you will receive the error message  “Not found merger for consumer name  bridgeSender'”  . The reason is that Magento does not know anything about this class. To declare this class (we are going to use it as Merger) you need to add the following code lines in the file <magento2ee_dir>/app/code/<module_name><etc>di.xml:

So now Magento knows which Merger class it should look for when you run the command:

You should not expect to see anything when the command is being implemented. The command will simply be listening to the queue and waiting for new messages. Moreover, if there occur any minor errors you will see nothing either, so you will not be aware of whether the script is running correctly or not. So to output error messages in the console I usually use the method getTransactionCallback in the file <magento2ee_dir>/vendor/magento/framework-message-queue/BatchConsumer.php:

So now, when we have Consumer available, let’s try to push a message to the queue. For that you need to add one component to the constructor class, as on the below example:

This code adds a message to the queue. Do not forget, that the object $order should be inherited from the interface that is indicated in the file queue.xml.

Now lets look at the method”MyModule\Consumer\ConsumerOrder”::”processMessage”

The $order parameter is a message that has already been parsed. By default magento deletes messaged that have been sent to consumer. But in case any error occurs and you need to repeat the attempt later, then you need to add the messages into the queue again.

So, to sum up, we can say that Magento 2.0 provides a very useful tool to work with queues – RabbitMq, which you can widely use in your modules.