Awesome Laravel: How to use RabbitMQ to handle communication between microservices

In this tutorial, we will see how we can exchange messages between two Laravel microservices using RabbitMQ. In practice, we will able to send a message from service A to service B to inform it something has happened, for example, a new user was created.

I will use Laravel but you can easily replicate it on every framework or on a pure PHP application. We will use the php-amqplib package.

You can see a pure PHP implementation here.

RabbitMQ installation

First, we need to install and run RabbitMQ on our local machine (or server). For example, you can install RabbitMQ on MacOS using homebrew (brew install rabbitmq) and run it executing brew services start rabbitmq.

How it works (in simple)

RabbitMQ is a message broker, so its goal is to handle messages exchange between multiple software. To satisfy this task, it provides messages queues and a TCP channel to receive messages from the producer and to have them read by the consumer. In practice, communication happens through this TCP channel.
In our case, Service A will be the producer (i.e. will send messages), and Service B will be the consumer (i.e. will read messages).

Project setup

We can now create two laravel projects and install the package in both projects.

composer require php-amqplib/php-amqplib

We have to define env variables to connect to RabbitMQ. So, add this snippet to your .env file (and also .env.example!) and change values if needed.

RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASS=guest

It is recommended to use config instead env to fetch environment data, so create config/rabbitmq.php and paste this

<?php

return [
    'host' => env('RABBITMQ_HOST', 'localhost'),
    'port' => env('RABBITMQ_PORT', 5672),
    'user' => env('RABBITMQ_USER', 'guest'),
    'pass' => env('RABBITMQ_PASS', 'guest'),
];

In this way, we can get the host in our Laravel application using config('rabbitmq.host');.

Step 1 — Connect to RabbitMQ

In order to connect to RabbitMQ, I will create in both projects a ConnectionAdapter class that takes connection parameters and starts a new connection.

Later we will need to get the connection channel and close the connection, so I have added these auxiliary functions.

<?php

namespace App;

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;

class ConnectionAdapter {
    private AMQPStreamConnection $connection;
    private AMQPChannel $channel;

    public function __construct() {
        $this->connection = new AMQPStreamConnection(
            config('rabbitmq.host'),
            config('rabbitmq.port'),
            config('rabbitmq.user'),
            config('rabbitmq.pass')
        );
        $this->channel = $this->connection->channel();
    }

    public function getChannel(): AMQPChannel {
        return $this->channel;
    }

    public function close(): void {
        $this->channel->close();
        $this->connection->close();
    }
}

Step 2 — Make the producer

Now we can focus only on Service A. We want to have a message (e.g. a string) and send it to the RabbitMQ queue.
If you want to handle JSON objects, you can use thejson_encode function.

We will create the MessageSender class in Service A. It will accept two parameters:

• The name of the queue (you can insert every name you want, RabbitMQ will handle this automatically);

• The previously made ConnectionAdapter.

The constructor will declare the queue (I created a trait for it). In fact, we declare to RabbitMQ the queue we are using and, if it doesn’t exist, RabbitMQ will create it.

We have two functions:

1. publish: sends the message to the RabbitMQ channel

2. sendMessage: our wrapper to send a message. It is useful because the producer should close the connection after he sends messages.

So the MessageProducer class will look like this:

<?php

namespace App;

use PhpAmqpLib\Message\AMQPMessage;

readonly class MessageProducer {
 use CanDeclareQueue;

 public function __construct(
  protected string            $queueName,
  protected ConnectionAdapter $connectionAdapter
 ) {
  $this->declareQueue();
 }

 protected function publish($messageBody): void {
  $message = new AMQPMessage($messageBody);
  // send the message to the queue (publish)
  $this->connectionAdapter->getChannel()->basic_publish($message, '', $this->queueName);
 }

 public function sendMessage($messageBody): void {
  $this->publish($messageBody);
  $this->connectionAdapter->close(); // close connection after the publish
 }
}

And this is the CanDeclareQueue trait:

<?php

namespace App;

trait CanDeclareQueue {
 protected function declareQueue(): void {
  $this->connectionAdapter->getChannel()->queue_declare($this->queueName, false, false, false, false);
 }
}

Now, we can use everywhere the following snippet to send a message:

$message = json_encode(['name' => 'Message!']); // it must be a string!
$producer = new MessageProducer('default', new ConnectionAdapter());
$producer->sendMessage($message); 

✅ Awesome! We sent a message using RabbitMQ! But… how can I read it?

Step 3 — Make the consumer

Now we can move on to the consumer. We can copy CanDetectQueue and ConnectionAdapter into Service B.

We want to receive messages and do something when we receive a new one. So, my idea is to create an artisan command php artisan rabbitmq:listen to start the consumer listening to the RabbitMQ queue and a MessageConsumer class to handle received messages (this operation is called consume).

Let’s start with the MessageConsumer class. We want to consume messages using a callback function and we want to wait for incoming messages.

To do this, we will make a receiveMessages function that accepts a callback and will consume messages sent to the communication channel.
The php-amqplib package has a function basic_consume we can execute on a channel to handle this.

So the MessageConsumer class will look like this:

<?php

namespace App;

readonly class MessageConsumer {
 use CanDeclareQueue;

 public function __construct(
  protected string            $queueName,
  protected ConnectionAdapter $connectionAdapter
 ) {
  $this->declareQueue();
 }

 public function receiveMessages(\Closure $callback): void {
  $this->consumeQueueMessages($callback);
  $this->connectionAdapter->close();
 }

 protected function consumeQueueMessages(\Closure $callback): void {
  // 2nd-6th parameters are queue options, in this tutorial we will ignore them
  // but read more on documentation
  $this->connectionAdapter->getChannel()
   ->basic_consume($this->queueName, '', false, true, false, false, $callback);
  $this->waitForMessages();
 }

 protected function waitForMessages(): void {
  // while channel is open
  while ($this->connectionAdapter->getChannel()->is_open()) {
   // wait for messages
   $this->connectionAdapter->getChannel()->wait();
  }
 }
}

We can now focus on the Artisan command.

<?php

namespace App\Console\Commands;

use App\ConnectionAdapter;
use App\MessageConsumer;
use Illuminate\Console\Command;

class RabbitMqListenCommand extends Command {
 protected $signature = 'rabbitmq:listen';

 public function handle() {
  // create a new consumer
  $consumer = new MessageConsumer('default', new ConnectionAdapter());
  $consumer->receiveMessages(function ($message) {
    // $message->body contains the value of the message sent by producer
    $this->info($message->body);
  });
 }
}

So we can now use the command php artisan rabbitmq:listen and wait for our messages!

✅ Today you learned how to handle a message broker between multiple Laravel projects. Clearly, you can use it to exchange data between different languages and frameworks!

If you liked this article, feel free to share on social media ;)