0

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?

2

This sounds like a good approach to me. You ask about this:

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.

Not in all cases. You can set a TTL both on the queue and on each message. If the consumer dies and nobody consumes the messages, the TTL configuration will make sure they are eventually deleted. See https://www.rabbitmq.com/ttl.html for details.

But I wonder if this is the correct behaviour from a business POV. If a consumer dies and a message is deleted due to TTL, the behaviour that this consumer implements for this event will never happen. No welcome email will be send. Is this intended?

RabbitMQ also supports automatic requeuing if the consumer fails to ACK a message. This could be helpful for recovering.

Is it okay to have lots of different consumers constantly running?

I don't know how PHP handles this. But the RabbitMQ server is fine as long as you don't open to many connections. Could it be possible to share a single connection to the RabbitMQ server between consumers? Then each consumer would only open a channel to the server using this connection, define the queue and the binding to the fanout exchange, and listen for messages. To reduce housekeeping work, you could use temporary queues that are removed as soon as a consumer disconnects or dies.

  • Thanks ! Sharing a connection makes a lot of sense actually. – Daniel Benzie May 15 at 10:55
  • On the subject of acknowledgements I was under the impression that if I don't send an acknowledgement the message will be re-queued ready for the consumer to come back online. e.g does this TTL thing mean basically it will be re-queued but it will be cleaned up eventually defined by a TTL time either on message or Q. EG for welcome email maybe I set it to 48 hours (makes no sense to send one after) – Daniel Benzie May 15 at 11:02
0

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.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.