RabbitMQ Consumer Architecture

Question

I have been investigating using RabbitMQ for a publish/subscribe pattern across micro-services where aysnc calls are appropriate but I am having trouble understanding some concepts and could not find any info in the manual.

I'll outline a basic example of how I wish to use it below.

I have a core service for user management. When a user is created I wish to publish an event such as user_created I would like this event to be consumed by multiple different services. For example

• A communications service which handles the sending of the welcome email
• A marketing service which takes the user data and publishes it to a third party data bucket
• A finance service which sets up a new customer statement/account

Each of the services above are their own separate codebase in a few different languages/frameworks.

From what I can tell of the docs for this kind of I will need an exchange where I publish a message like user_created along with any data I wish to send (like the user object)

On each of my services I then set up a consumer script which connects to the exchange and creates a queue and consumes any messages sent.

If my understanding so far is correct my main question comes from how to handle these consumers on the services end. If I am using acknowledgement my understanding is that even if one of my consumers dies the queue and message will not be deleted. Both will persist and wait for the a new consumer indefinitely.

Any number of services will presumably want to consume lots of different messages. Is it okay to have lots of different consumers constantly running?

Primarily the services use PHP so these consumers would maybe look like the below (for the comms service)

sendWelcomeEmail.php
sendGoodbyeEmail.php
sendInvoice.php

So presumably I set up each of these scripts as directed on the tutorials (like here https://www.rabbitmq.com/tutorials/tutorial-three-php.html) and run the scripts as part of the deployment process. I would then need something like supervisor to make sure the consumer is running and if it is not restart it. This could be quite cumbersome having to set a supervisor up for each consumer. Are there better ways to handle things?

Answer 1

If my understanding so far is correct my main question comes from how to handle these consumers on the services end. If I am using acknowledgement my understanding is that even if one of my consumers dies the queue and message will not be deleted. Both will persist and wait for the a new consumer indefinitely.

Yes, you should use an exchange publishing to one queue per use case, as described here.

For making sure that a consumer actually receives and can handle a message, read up on consumer acknowledgements here.

If you need the queue to keep not consumed messages even if rabbitmq dies, use persistent queues (scroll down to Message durability):

Any number of services will presumably want to consume lots of different messages. Is it okay to have lots of different consumers constantly running?

Yes that's pretty much what it's designed for. If rabbitMQ gets slow at some point, you can also scale rabbitMQ, but I cannot talk about any experience for the reliability here, as I have never had to use the scaling.

One other question that you should keep in mind when working with a message broker is how you want to introduce a new service which needs to consume the same messages (in this case users). Due to the exchange only routing messages and not persisting them - the messages sent before will be lost, so you will need a different way to populate new services.

Answer 2

Here is what I am using in my project. I was so very confused and when I figured it out it was so simple. My project is still very much in it's infancy but this part seems to work great so far.

All of my consumers run together in one loop and are dispatched perfectly. One connection and one channel.

I configure the queues and the function to handle them like this:

'queues' => [
                [
                    'name' => 'q_command_order_create',
                    'callback' => [
                        AMQPCommandBridge::class,
                        'process_message_create_new_order'
                    ]
                ],
                [
                    'name' => 'q_command_order_addsalesline',
                    'callback' => [
                        AMQPCommandBridge::class,
                        'process_message_addsalesline'
                    ]
                ],
                [
                    'name' => 'q_command_order_addserviceline',
                    'callback' => [
                        AMQPCommandBridge::class,
                        'process_message_addserviceline'
                    ]
                ],
// Etc...

The config has the queue name for the consumer and the callback as Class and function within the class. I have one class (AMQPCommandBridge) right now that handles all of my consumer callbacks but I will likely split them up more appropriately as the system matures.

Here is an example function in my AMQPCommandBridge class:

public function process_message_create_new_order($message)
{
    // First just a debug message
    echo $message->delivery_info['routing_key'] . PHP_EOL;

    // Send a command to the domain
    $orderId = json_decode($message->body)->orderId;
    $this->commandBus->handle(new OrderCreateNewCommand($orderId));

    // Acknowledge the receipt 
    $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
}

I start all of the consumer(s) like this in my main.php file:

echo 'Starting RabbitMQ consumer...' . PHP_EOL;
flush();

$rabbitConsumer = new RabbitConsumer($container);
$rabbitConsumer->run();

Here is my consumer class that configures the consumers from my config and sets the callbacks.

<?php

declare(strict_types=1);

namespace Micro\Infrastructure;


use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exception\AMQPRuntimeException;
use Psr\Container\ContainerInterface;

//use Micro\Service\AMQPShutdownService; // ?!?!


class RabbitConsumer {

    /* @var ContainerInterface */
    private $container;

    /* @var AMQPStreamConnection */
    private $connection;

    /* @var array */
    private $queues_array;

    const WAIT_BEFORE_RECONNECT_uS = 3000000;


    public function __construct(ContainerInterface $container)
    {
        $this->container = $container;
        $this->connection = null;

        $this->queues_array = $container->get('config')['settings']['amqp']['queues'];
    }

    public function run()
    {
        while(true) {
            try {
                // Connect to the AMQP service
                $this->connection = $this->container->get(AMQPStreamConnection::class);

                // Register the proper shutdown of the connection so queue's will drop
//                register_shutdown_function(array(AMQPShutdownService::class, 'shutdown', $this->connection));

                // Start the consumer
                $this->consume();

            } catch(AMQPRuntimeException $e) {
                echo $e->getMessage() . PHP_EOL;
                $this->cleanup_connection();
                usleep(self::WAIT_BEFORE_RECONNECT_uS);
            } catch(\RuntimeException $e) {
                echo "Runtime exception: " . $e->getMessage(). PHP_EOL;
                $this->cleanup_connection();
                usleep(self::WAIT_BEFORE_RECONNECT_uS);
            } catch(\InvalidArgumentException $e) {
                echo "Invalid argument exception: " . $e->getMessage() . PHP_EOL;
//                $this->cleanup_connection();
                usleep(self::WAIT_BEFORE_RECONNECT_uS);
            } catch(\ErrorException $e) {
                echo "Error exception: " . $e->getMessage() . PHP_EOL;
                $this->cleanup_connection();
                usleep(self::WAIT_BEFORE_RECONNECT_uS);
            }
        }
    }

    protected function cleanup_connection() {
        // Connection might already be closed.
        // Ignoring exceptions.
        try {
            if($this->connection !== null) {
                $this->connection->close();
            }
        } catch (\Exception $e) {
            // ignore the exception
        }
    }


    protected function consume()
    {
        $channel = $this->connection->channel();

        foreach($this->queues_array as $queue) {
            $channel->queue_declare(
                $queue['name'],
                false,
                true,
                false,
                false
            );

            $consumeClass = $this->container->get($queue['callback'][0]);

            $channel->basic_consume(
                $queue['name'],
                'consumer_' . $queue['name'],
                false,
                false,
                false,
                false,
                [$consumeClass, $queue['callback'][1]]
            );
        }

        while ($channel->is_consuming()) {
            $channel->wait();
        }
    }

}

I hope this helps and is not too confusing. I am learning a ton about this stuff right now. Happy to help others.