1

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?

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.

| improve this answer | |

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.