In part 1 we covered the theory and a simple use case of the AMQP protocol in PHP with RabbitMQ as the broker. Now, let’s dive into some more advanced examples.
Example 1: send request to process data asynchronously among several workers
In the example of the previous part, we had one producer, one consumer. If the consumer died, messages would continue to stack up in the queue until the consumer started again. It would then process all the messages, one by one.
This can be less than ideal in a concurrent user environment with a fair amount of requests per minute. Fortunately, scaling the consumers is super easy, but let’s implement another example.
Let’s say we have an invoice generation service, where the users just need to provide the invoice number, and the system will automatically generate a pdf file and email it to the user. Generating and sending the email could take even several seconds if the server on which the generation process runs is resource limited. Now suppose we are required to support several transactions per second, how do we accomplish this without overwhelming the server?
We need to implement the following pattern:
Let’s look at our producer class:
<?php
namespace Acme\AmqpWrapper;
use PhpAmqpLib\Connection\AMQPConnection;
use PhpAmqpLib\Message\AMQPMessage;
class WorkerSender
{
/* ... SOME OTHER CODE HERE ... */
/**
* Sends an invoice generation task to the workers
*
* @param int $invoiceNum
*/
public function execute($invoiceNum)
{
$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare(
'invoice_queue', #queue - Queue names may be up to 255 bytes of UTF-8 characters
false, #passive - can use this to check whether an exchange exists without modifying the server state
true, #durable, make sure that RabbitMQ will never lose our queue if a crash occurs - the queue will survive a broker restart
false, #exclusive - used by only one connection and the queue will be deleted when that connection closes
false #auto delete - queue is deleted when last consumer unsubscribes
);
$msg = new AMQPMessage(
$invoiceNum,
array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
);
$channel->basic_publish(
$msg, #message
'', #exchange
'invoice_queue' #routing key (queue)
);
$channel->close();
$connection->close();
}
}
The WorkerSender::execute()
method will receive an invoice number. Next we create a connection, channel and queue as usual.
<?php
/* ... SOME CODE HERE ... */
$msg = new AMQPMessage(
$invoiceNum,
array('delivery_mode' => 2) # make message persistent, so it is not lost if server crashes or quits
);
/* ... SOME CODE HERE ... */
Please notice that this time, while creating the message object, the constructor receives a second parameter: array('delivery_mode' => 2)
. In this case we want to state that the message should not be lost if the RabbitMQ server crashes. Please be aware that in order for this to work, the queue has to be declared durable, too.
The following code can be used to receive the form data and execute the producer:
<?php
chdir(dirname(__DIR__));
require_once('vendor/autoload.php');
use Acme\AmqpWrapper\WorkerSender;
$inputFilters = array(
'invoiceNo' => FILTER_SANITIZE_NUMBER_INT,
);
$input = filter_input_array(INPUT_POST, $inputFilters);
$sender = new WorkerSender();
$sender->execute($input['invoiceNo']);
Please use whichever input sanitization/validation you feel comfortable with.
Things get a little bit interesting on the consumer side:
<?php
namespace Acme\AmqpWrapper;
use PhpAmqpLib\Connection\AMQPConnection;
use PhpAmqpLib\Message\AMQPMessage;
class WorkerReceiver
{
/* ... SOME OTHER CODE HERE ... */
/**
* Process incoming request to generate pdf invoices and send them through
* email.
*/
public function listen()
{
$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare(
'invoice_queue', #queue
false, #passive
true, #durable, make sure that RabbitMQ will never lose our queue if a crash occurs
false, #exclusive - queues may only be accessed by the current connection
false #auto delete - the queue is deleted when all consumers have finished using it
);
/**
* don't dispatch a new message to a worker until it has processed and
* acknowledged the previous one. Instead, it will dispatch it to the
* next worker that is not still busy.
*/
$channel->basic_qos(
null, #prefetch size - prefetch window size in octets, null meaning "no specific limit"
1, #prefetch count - prefetch window in terms of whole messages
null #global - global=null to mean that the QoS settings should apply per-consumer, global=true to mean that the QoS settings should apply per-channel
);
/**
* indicate interest in consuming messages from a particular queue. When they do
* so, we say that they register a consumer or, simply put, subscribe to a queue.
* Each consumer (subscription) has an identifier called a consumer tag
*/
$channel->basic_consume(
'invoice_queue', #queue
'', #consumer tag - Identifier for the consumer, valid within the current channel. just string
false, #no local - TRUE: the server will not send messages to the connection that published them
false, #no ack, false - acks turned on, true - off. send a proper acknowledgment from the worker, once we're done with a task
false, #exclusive - queues may only be accessed by the current connection
false, #no wait - TRUE: the server will not respond to the method. The client should not wait for a reply method
array($this, 'process') #callback
);
while(count($channel->callbacks)) {
$this->log->addInfo('Waiting for incoming messages');
$channel->wait();
}
$channel->close();
$connection->close();
}
/**
* process received request
*
* @param AMQPMessage $msg
*/
public function process(AMQPMessage $msg)
{
$this->generatePdf()->sendEmail();
/**
* If a consumer dies without sending an acknowledgement the AMQP broker
* will redeliver it to another consumer or, if none are available at the
* time, the broker will wait until at least one consumer is registered
* for the same queue before attempting redelivery
*/
$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
}
/**
* Generates invoice's pdf
*
* @return WorkerReceiver
*/
private function generatePdf()
{
/**
* Mocking a pdf generation processing time. This will take between
* 2 and 5 seconds
*/
sleep(mt_rand(2, 5));
return $this;
}
/**
* Sends email
*
* @return WorkerReceiver
*/
private function sendEmail()
{
/**
* Mocking email sending time. This will take between 1 and 3 seconds
*/
sleep(mt_rand(1,3));
return $this;
}
}
As usual, we have to create a connection, derive a channel and declare a queue (the queue’s parameters have to be de same as the producer).
<?php
/* ... SOME CODE HERE ... */
$channel->basic_qos(
null, #prefetch size - prefetch window size in octets, null meaning "no specific limit"
1, #prefetch count - prefetch window in terms of whole messages
null #global - global=null to mean that the QoS settings should apply per-consumer, global=true to mean that the QoS settings should apply per-channel
);
/* ... SOME CODE HERE ... */
In order to have worker behavior (dispatch messages among several proceses), we have to declare the Quality of Service (qos) parameters with $channel->basic_qos()
:
- Prefetch size: no specific limit, we could have as many workers as we need
- Prefetch count: how many messages to retrieve per worker before sending back an acknowledgement. This will make the worker to process 1 message at a time.
- Global: null means that above settings will apply to this consumer only
Next, we will start consuming, with a key difference in the parameters. We will turn off automatic ack’s, since we will tell the RabbitMQ server when we have finished processing the message and be ready to receive a new one.
Now, how do we send that ack? Please take a look at the WorkerReceiver::process()
method (which is declared as a callback method when a message is received). The calls to generatedPdf()
and sendEmail()
methods are just dummy methods that will simulate the time spent to accomplish both tasks. The $msg
parameter not only contains the payload sent from the producer, it also contains information about the objects used by the producer. We can extract information about the channel used by the producer with $msg->delivery_info['channel']
(which is the same object type to the channel we opened for the consumer with $connection->channel();
). Since we need to send the producer’s channel an acknowledgement that we have completed the process, we will use its basic_ack()
method, sending as a parameter the delivery tag ($msg->delivery_info['delivery_tag']
) RabbitMQ automatically generated in order to associate correctly to which message the ack belongs to.
How do we fire up the workers? Just create a file like the following, invoking the WorkerReceiver::listen()
method:
<?php
chdir(dirname(__DIR__));
require_once('vendor/autoload.php');
use Acme\AmqpWrapper\WorkerReceiver;
$worker = new WorkerReceiver();
$worker->listen();
Now use the php command (e.g. php worker.php
or whichever name you have given to above file) to fire up the worker. But wait, the purpose was to have two or more workers, wasn’t it? No problem, fire up more workers in the same way in order to have multiple processes of the same file, and RabbitMQ will register the consumers and distribute work among them according to the QoS parameters.
Example 2: send RPC requests and expect a reply
So far, we have been sending messages to the RabbitMQ server without the user having to wait for a reply. This is ok for asynchronous processes that might take more time than the user is willing to spend just to see an ‘OK’ message. But what if we actually need a reply? Let’s say some result from a complex calculation, so we can show it to the user?
Let’s say we have a centralized login server (single sign on) that will work as an authentication mechanism isolated from the rest of our application(s). The only way to reach this server is through RabbitMQ. We need to implement a way to send the login credentials to this server and wait for a grant/deny access response.
We need to implement the following pattern:
As usual, let’s look at the producer first:
<?php
namespace Acme\AmqpWrapper;
use PhpAmqpLib\Connection\AMQPConnection;
use PhpAmqpLib\Message\AMQPMessage;
class RpcSender
{
private $response;
/**
* @var string
*/
private $corr_id;
/* ...SOME OTHER CODE HERE... */
/**
* @param array $credentials
* @return string
*/
public function execute($credentials)
{
$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
/*
* creates an anonymous exclusive callback queue
* $callback_queue has a value like amq.gen-_U0kJVm8helFzQk9P0z9gg
*/
list($callback_queue, ,) = $channel->queue_declare(
"", #queue
false, #passive
false, #durable
true, #exclusive
false #auto delete
);
$channel->basic_consume(
$callback_queue, #queue
'', #consumer tag
false, #no local
false, #no ack
false, #exclusive
false, #no wait
array($this, 'onResponse') #callback
);
$this->response = null;
/*
* $this->corr_id has a value like 53e26b393313a
*/
$this->corr_id = uniqid();
$jsonCredentials = json_encode($credentials);
/*
* create a message with two properties: reply_to, which is set to the
* callback queue and correlation_id, which is set to a unique value for
* every request
*/
$msg = new AMQPMessage(
$jsonCredentials, #body
array('correlation_id' => $this->corr_id, 'reply_to' => $callback_queue) #properties
);
/*
* The request is sent to an rpc_queue queue.
*/
$channel->basic_publish(
$msg, #message
'', #exchange
'rpc_queue' #routing key
);
while(!$this->response) {
$channel->wait();
}
$channel->close();
$connection->close();
return $this->response;
}
/**
* When a message appears, it checks the correlation_id property. If it
* matches the value from the request it returns the response to the
* application.
*
* @param AMQPMessage $rep
*/
public function onResponse(AMQPMessage $rep) {
if($rep->get('correlation_id') == $this->corr_id) {
$this->response = $rep->body;
}
}
}
Looking at RpcSender::execute
method, please be aware that the $credentials
parameter is an array in the form of ['username'=>'x', 'password'=>'y']
. Again, we open a new connection and create a channel as usual.
<?php
//...
list($callback_queue, ,) = $channel->queue_declare(
"", #queue
false, #passive
false, #durable
true, #exclusive
false #auto delete
);
//...
The first difference comes from declaring a queue. First notice that we are using the list()
construct to catch the result from $channel->queue_declare()
. This is because we do not explicitly send a queue name while declaring it so we need to find out how this queue is identified. We are only interested in the first element of the result array, which will be an unique identifier of the queue (something like amq.gen-_U0kJVm8helFzQk9P0z9gg
). The second change is that we need to declare this queue as exclusive, so there is no mix up in the results from other concurrent processes.
Another big change is that the producer will be a consumer of a queue too, when executing $channel->basic_consume()
please notice that we are providing the $callback_queue
value we got while declaring the queue. And like every consumer, we will declare a callback to execute when the process receives a response.
<?php
//...
/*
* $this->corr_id has a value like 53e26b393313a
*/
$this->corr_id = uniqid();
//...
Next, we have to create a correlation id for the message, this is nothing more than a unique identifier for each message. In the example we are using uniqid()
’s output, but you can use whichever mechanism you prefer (as long as it does not create a race condition, does not need to be a strong, crypto-safe RNG).
<?php
//...
$msg = new AMQPMessage(
$jsonCredentials, #body
array('correlation_id' => $this->corr_id, 'reply_to' => $callback_queue) #properties
);
//...
Now let’s create a message, which has important changes compared to what we were used to in the first 2 examples. Aside from assigning a json-encoded string containing the credentials we want to authenticate, we have to provide to the AMQPMessage
constructor an array with two properties defined:
- correlation_id: a tag for the message
- reply_to: the queue identifier generated while declaring it
After publishing the message, we will evaluate the response, which will be empty at the beginning. While the response value remains empty, we will wait for a response from the channel with $channel->wait();
.
Once we receive a response from the channel, the callback method will be invoked (RpcSender::onResponse()
). This method will match the received correlation id against the one generated, and if they are the same, will set the response body, thus breaking the while loop.
What about the RPC consumer? Here it is:
<?php
namespace Acme\AmqpWrapper;
use PhpAmqpLib\Connection\AMQPConnection;
use PhpAmqpLib\Message\AMQPMessage;
class RpcReceiver
{
/* ... SOME OTHER CODE HERE... */
/**
* Listens for incoming messages
*/
public function listen()
{
$connection = new AMQPConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare(
'rpc_queue', #queue
false, #passive
false, #durable
false, #exclusive
false #autodelete
);
$channel->basic_qos(
null, #prefetch size
1, #prefetch count
null #global
);
$channel->basic_consume(
'rpc_queue', #queue
'', #consumer tag
false, #no local
false, #no ack
false, #exclusive
false, #no wait
array($this, 'callback') #callback
);
while(count($channel->callbacks)) {
$channel->wait();
}
$channel->close();
$connection->close();
}
/**
* Executes when a message is received.
*
* @param AMQPMessage $req
*/
public function callback(AMQPMessage $req) {
$credentials = json_decode($req->body);
$authResult = $this->auth($credentials);
/*
* Creating a reply message with the same correlation id than the incoming message
*/
$msg = new AMQPMessage(
json_encode(array('status' => $authResult)), #message
array('correlation_id' => $req->get('correlation_id')) #options
);
/*
* Publishing to the same channel from the incoming message
*/
$req->delivery_info['channel']->basic_publish(
$msg, #message
'', #exchange
$req->get('reply_to') #routing key
);
/*
* Acknowledging the message
*/
$req->delivery_info['channel']->basic_ack(
$req->delivery_info['delivery_tag'] #delivery tag
);
}
/**
* @param \stdClass $credentials
* @return bool
*/
private function auth(\stdClass $credentials) {
if (($credentials->username == 'admin') && ($credentials->password == 'admin')) {
return true;
} else {
return false;
}
}
}
Same old connection and channel creation :)
Same as declaring the queue, however this queue will have a predefined name (‘rpc_queue‘). We will define the QoS parameters since we will deactivate automatic acks, so we can notify when we are finished verifying the credentials and have a result.
<?php
//...
$msg = new AMQPMessage(
json_encode(array('status' => $authResult)), #message
array('correlation_id' => $req->get('correlation_id')) #options
);
//...
The magic comes from within the declared callback. Once we are done authenticating the credentials (yes, I know the process is done against static username/password values, this tutorial is not about how to authenticate credentials ;) ), we have to create the return message with the same correlation id the producer created. We can extract this from the request message with $req->get('correlation_id')
, passing this value the same way we did it in the producer.
<?php
//...
$req->delivery_info['channel']->basic_publish(
$msg, #message
'', #exchange
$req->get('reply_to') #routing key
);
//...
Now we have to publish this message to the same queue that was created in the producer (the one with the ‘random’ name). We extract the queue’s name with $req->get('reply_to')
and use it as the routing key in basic_publish()
.
Once we published the message, we have to send the ack notice to the channel with $req->delivery_info['channel']->basic_ack()
, using the delivery tag in $req->delivery_info['delivery_tag']
so the producer can stop waiting.
Again, fire up a listening process and you are ready to go. You can even combine examples 2 and 3 to have a multi-worker rpc process to perform the authentication requests than can be scaled just by firing up several workers.
There is far more to be said about RabbitMQ and AMQP, like virtual hosts, exchange types, server administration, etc… you can find more application patterns (like routing, topics) here and at the documentation page. There is also a command line tool to manage RabbitMQ, as well as a web based interface.
If you liked this tutorial series and would like to see more about MQ and more real world use cases, please let us know in the comments below!
Frequently Asked Questions (FAQs) about PHP RabbitMQ Advanced Examples
What is the role of RabbitMQ in PHP?
RabbitMQ is a message broker that allows applications to communicate with each other asynchronously. It plays a crucial role in PHP applications by enabling them to handle high loads and complex tasks more efficiently. RabbitMQ uses the Advanced Message Queuing Protocol (AMQP) to facilitate the exchange of messages between different parts of an application. This allows for the decoupling of processes, making the application more scalable and resilient.
How do I install RabbitMQ for PHP?
To install RabbitMQ for PHP, you need to first install RabbitMQ server on your machine. This can be done through the official RabbitMQ website. After the server is installed, you can install the PHP AMQP extension, which provides the necessary functions to interact with RabbitMQ. This can be done using the PECL installer with the command pecl install amqp
.
How can I create a RabbitMQ exchange in PHP?
In PHP, you can create a RabbitMQ exchange using the exchange_declare
method of the AMQPChannel
class. This method takes several parameters, including the name of the exchange, the type of the exchange (direct, topic, fanout, or headers), and optional parameters such as passive, durable, auto_delete, and arguments.
How do I send a message to a RabbitMQ queue in PHP?
To send a message to a RabbitMQ queue in PHP, you first need to create an instance of the AMQPMessage
class with the message content. Then, you can use the basic_publish
method of the AMQPChannel
class to send the message to the queue. The basic_publish
method takes the message, the exchange, and the routing key as parameters.
How can I consume messages from a RabbitMQ queue in PHP?
In PHP, you can consume messages from a RabbitMQ queue using the basic_consume
method of the AMQPChannel
class. This method takes several parameters, including the queue name, the consumer tag, no_local, no_ack, exclusive, and a callback function that will be executed when a message is received.
How can I handle errors in RabbitMQ with PHP?
Error handling in RabbitMQ with PHP can be done using try-catch blocks. The PHP AMQP extension throws exceptions of the AMQPException
class when an error occurs. You can catch these exceptions and handle them according to your application’s needs.
How can I ensure message durability in RabbitMQ with PHP?
To ensure message durability in RabbitMQ with PHP, you can set the delivery_mode
property of the AMQPMessage
class to 2. This will make RabbitMQ store the message on disk, ensuring that it will not be lost even if the RabbitMQ server crashes or restarts.
How can I implement priority queues in RabbitMQ with PHP?
Priority queues in RabbitMQ can be implemented in PHP by setting the x-max-priority
argument when declaring the queue. Then, when sending a message, you can set the priority
property of the AMQPMessage
class to a value between 0 and the maximum priority you specified.
How can I use RabbitMQ for RPC in PHP?
RabbitMQ can be used for Remote Procedure Call (RPC) in PHP by sending a message with a reply-to property set to a callback queue. The server can then send the response to the callback queue, and the client can consume the response from there.
How can I monitor RabbitMQ in PHP?
Monitoring RabbitMQ in PHP can be done using the RabbitMQ management plugin, which provides a web-based interface for monitoring and managing your RabbitMQ server. You can also use the AMQPChannel
class’s methods to get information about the state of the channel, such as the number of unacknowledged messages.
![Miguel Ibarra Romero](https://uploads.sitepoint.com/wp-content/uploads/2014/01/miguel_romero-96x96.png)
Web application developer, database administrator, project lead in a variety of enterprise apps and now article author. Interested in database design and optimization. Amazed and involved in distributed systems development. Cryptography and information security fan.