PHP

PHP and RabbitMQ: Advanced Examples

Miguel Ibarra Romero
Share

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!