Symfony + RabbitMQ Quick Start for Young

Good day, friends.
Today I wanted to talk about how you can work with RabbitMQ in Symfony and very little about some underwater combs. In the end, I will write a couple of interesting moments about the rabbit (Russian translation of "rabbit") for those who are completely in the tank.

I won’t talk about RabbitMQ itself, so if you don’t even know this yet, read the following translations:

Article 1
Article 2
Article 3
Article 4
Article 5

Don't be afraid of examples on pearl or python - it's not scary, everything is clear enough from the source code.

+ Everything is described in sufficient detail when I read it in due time, it was enough to interpret the code mentally to understand how what and why.

If you already know what a consumer is and why you need to do $ em-> clear () + gc_collect_cycles in it, and then close the database connection, then most likely you will not learn anything new for yourself. The article is more likely for those who do not want to deal with the AMQP protocol, but who need to apply the queues right now and for some reason the choice fell mindlessly on RabbitMQ, and not the same lightweight beanstalkd .
If you have a microservice architecture and you expect me to tell you how to weld communication between components through AMQP, how to make RPC beautifully, then I myself have been waiting for something like this for a very long time ...

We have a task: to send messages to Email in a queue using RabbitMQ, as well as provide fault tolerance: if the mail server answered with a timeout or something else broke, you need to try to complete the task again after 30 seconds.

So, install our bundle .

I'm too lazy to tell you how to copy the composer require command and line in AppKernel.

I really hope that you yourself have done this and are ready to start configuring our bundle.



Bundle configuration for us:

old_sound_rabbit_mq:
    connections:
        default:
            host:     'localhost'
            port:     5672
            user:     'guest'
            password: 'guest'
            vhost:    '/'
            lazy:     false
            connection_timeout: 3
            read_write_timeout: 3
            keepalive: false
            heartbeat: 0
            use_socket: true
    producers:
        send_email:
            connection:       default
            exchange_options: { name: 'notification.v1.send_email', type: direct }
    consumers:
        send_email:
            connection:       default
            exchange_options: { name: 'notification.v1.send_email', type: direct }
            queue_options:    { name: 'notification.v1.send_email' }
            callback:         app.consumer.mail_sender

Here, great attention should be paid to producers and consumers. If it is very short and simple: producer is what sends messages through RabbitMQ to consumer, and consumer, in turn, is the thing that receives and processes these messages. Here exchange_options are options for the exchanger (did you read articles about rabbitmq that were at the beginning of the article?), Queue_options are options for the queue (similarly). It is also worth paying attention to the callback in consumer - here is the service ID that extends the ConsumerInterface (execute method with a message argument).

Because so far you don’t have it, when you start the application or compile the container, we get some DI exception that the service was not found, but we request it. So let's create our service:

#app/config/services.yml
services:
    app.consumer.mail_sender:
        class: AppBundle\Consumer\MailSenderConsumer 

And the class itself:

namespace AppBundle\Consumer;
use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;
/**
 * Class NotificationConsumer
 */
class MailSenderConsumer implements ConsumerInterface
{
    /**
     * @var AMQPMessage $msg
     * @return void
     */
    public function execute(AMQPMessage $msg)
    {
        echo 'Ну тут типа сообщение пытаюсь отправить: '.$msg->getBody().PHP_EOL;
        echo 'Отправлено успешно!...';
    }
}

Well, you were not offended that I did not include in the article how to work with SwiftMailer? :) It is important for us that a string is delivered here asynchronously through a message queue, the way we process this string is our business. Mail is just an example of a case.

How do we pass the string to our consumer? To do this, let's create a test command:

namespace AppBundle\Command;
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
class TestConsumerCommand extends ContainerAwareCommand
{
    /**
     * {@inheritdoc}
     */
    protected function configure()
    {
        $this
            ->setName('app:test-consumer')
            ->setDescription('Hello PhpStorm');
    }
    /**
     * {@inheritdoc}
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $this->getContainer()->get('old_sound_rabbit_mq.send_email_producer')->publish('Сообщенька для отправки на мыло...');
    }

Again, I apologize that I did not make a controller with a beautiful mold for you - I'm too lazy for this. Yes, and too much. But I’m an economical lazy person and I like to draw, I dream a little towards theories and application architecture. Distracted.

Now we launch our consumer and order him to wait for messages from RabbitMQ:

bin/console rabbitmq:consumer send_email -vvv

And send him a message from our test team:

bin/console app:test-consumer

And now, in the process of rabbitmq: consumer, we can see our message! And that pseudo dispatch was a success.

Now let's see how delayed message processing can be implemented in case of errors. I will not use the RabbitMQ plugin for pending messages. We will achieve this by creating a new queue, in which we indicate the message lifetime 30 seconds and set the setting: after death - transferred to the main queue.

Just add a new producer:

producers:
        send_email:
            connection:       default
            exchange_options: { name: 'notification.v1.send_email', type: direct }
        delayed_send_email:
            connection: default
            exchange_options:
                name: 'notification.v1.send_email_delayed_30000'
                type: direct
            queue_options:
                name: 'notification.v1.send_email_delayed_30000'
                arguments:
                    x-message-ttl: ['I', 30000]
                    x-dead-letter-exchange: ['S', 'notification.v1.send_email']

Now let's change the logic of the consumer:

namespace AppBundle\Consumer;
use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface;
use PhpAmqpLib\Message\AMQPMessage;
/**
 * Class NotificationConsumer
 */
class MailSenderConsumer implements ConsumerInterface
{
    private $delayedProducer;
    /**
     * MailSenderConsumer constructor.
     * @param ProducerInterface $delayedProducer
     */
    public function __construct(ProducerInterface $delayedProducer)
    {
        $this->delayedProducer = $delayedProducer;
    }
    /**
     * @var AMQPMessage $msg
     * @return void
     */
    public function execute(AMQPMessage $msg)
    {
        $body = $msg->getBody();
        echo 'Ну тут типа сообщение отправляю '.$body.' ...'.PHP_EOL;
        try {
            if ($body == 'bad') {
                throw new \Exception();
            }
            echo 'Успешно отправлено...'.PHP_EOL;
        } catch (\Exception $exception) {
            echo 'ERROR'.PHP_EOL;
            $this->delayedProducer->publish($body);
        }
    }
}

In general, it is useful to use LoggerInterface for output - and it is beautiful and scalable.
But we are too lazy and we do not want to create additional “thoughts”, right? Just know.

Now we have to skip producer for the deferred queue:

#app/config/services.yml
services:
    app.consumer.mail_sender:
        class: AppBundle\Consumer\MailSenderConsumer
        arguments: ['@old_sound_rabbit_mq.delayed_send_email_producer']

And change the command:

namespace AppBundle\Command;
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
class TestConsumerCommand extends ContainerAwareCommand
{
    /**
     * {@inheritdoc}
     */
    protected function configure()
    {
        $this
            ->setName('app:test-consumer')
            ->setDescription('Hello PhpStorm');
    }
    /**
     * {@inheritdoc}
     */
    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $this->getContainer()->get('old_sound_rabbit_mq.send_email_producer')->publish('Ура, сообщенька...');
        $this->getContainer()->get('old_sound_rabbit_mq.send_email_producer')->publish('bad');
    }
}

Now, together with a normal message, she will send a bad message.

If we run, we will see the following output:

Ну тут типа сообщение отправляю Ура, сообщенька...
Успешно отправлено...
Ну тут типа сообщение отправляю bad...
ERROR

After 30 seconds, a processing message appears again:

Ну тут типа сообщение отправляю bad...
ERROR

And so endlessly. The logic of maximum attempts, etc. think for yourself. Next, I will give a couple of tips for your sales and some features.

Now tips for your sales:

1) Without leaving the topic with maximum processing attempts: be aware of all 102% of all possible exceptions to the context with which you are working! Learn to imagine when re-processing is required, and when not, otherwise - hello to the trash from the logs and lack of understanding of what happens. If the beaten task is spinning in RabbitMQ, with real data, normal tasks, you are unlikely to be able to throw out broken tasks without crutches without updating the code of the consumer and restarting it. So think it over right away. In this case, it would be correct to catch only some SMTPTimeOutException.
Also with such a model, it is important to understand that: on the first stage - one “global responsibility for changing the state of something”. Do not give too many risky tasks to your worker. If we consider the option with 1C, then the problem may be as follows: for example, if we successfully or unsuccessfully change \ add the product to 1C, we write something to the database, for example, the date of the last successful synchronization or unsuccessful. Those. here 2 databases are updated at once: 1C database and your application database. Suppose everything was successfully created in 1C, then the field “date of the last successful synchronization” is updated in the database - hop, an error popped up, again, the database server does not respond - the task is postponed to “later” and repeated until the database begins to respond. And at the same time, each time the "subtask" associated with creating an entity in 1C will be successfully performed,

2) Read about durable, since we are using RabbitMQ. PS: it starts up as true \ false the “durable” flag in the bundle config, specifically in exchange_options and queue_options

3) All your life, close the database connection after the program has been executed. And also run EM cleanup and after the garbage collector to clean up links. Those. in the end, our consumer should look something like this:

class MailSenderConsumer implements ConsumerInterface
{
    private $delayedProducer;
    private $entityManager;
    /**
     * MailSenderConsumer constructor.
     * @param ProducerInterface      $delayedProducer
     * @param EntityManagerInterface $entityManager
     */
    public function __construct(ProducerInterface $delayedProducer, EntityManagerInterface $entityManager)
    {
        $this->delayedProducer = $delayedProducer;
        $this->entityManager = $entityManager;
        gc_enable();
    }
    /**
     * @var AMQPMessage $msg
     * @return void
     */
    public function execute(AMQPMessage $msg)
    {
        $body = $msg->getBody();
        echo 'Ну тут типа сообщение отправляю '.$body.' ...'.PHP_EOL;
        try {
            if ($body == 'bad') {
                throw new \Exception();
            }
            echo 'Успешно отправлено...'.PHP_EOL;
        } catch (\Exception $exception) {
            echo 'ERROR'.PHP_EOL;
            $this->delayedProducer->publish($body);
        }
        $this->entityManager->clear();
        $this->entityManager->getConnection()->close();
        gc_collect_cycles();
    }
}

The consumer works like a daemon, so constantly storing links in it and keeping a connection to the database is bad. In the case of MySQL, you get MySQL server has gone away.

4) Think a lot about why your pending message model can unexpectedly kill your business. For example, we have a mechanism that, when changing a product in the admin panel, pours these changes through the queue in 1C. Now imagine the situation: the administrator changes the product -> task # 1 is created to try to change the same data in the 1C database. The 1C server does not respond, so the task simply shifts constantly until everything works. During this time, the administrator decided to fix something else in the same product, which he does. Task # 2 is logged.

Now imagine a situation where tasks # 1 and # 2 are executed and postponed one by one.

What if 1C is working by the time task # 2 is completed? The task will be completed and flood the latest changes. Next, task # 1 will be used and stable changes will be erased :)
Exit: send timestamp as version, and if the task is “from the past”, we throw it away.

5) You go into asynchrony - read about many architectural problems, as well as race condition, inconsistency of consumers on different machines and more.

6) Write the versions to your queues ... Wow, how it helps on a real prod. In principle, we did just that in this example.

7) Perhaps you do not need RabbitMQ and the whole AMQP protocol. Look towards beanstalkd.

8) Run the consumers and everything else demonic on php through supervisor and connect the full logging of the fall of processes in it. He also has a web interface to manage all this business, which is also very convenient. There will always be problems.