Distributed queue example

There were lots of discussing lately about implementing distributed message queues using MySQL.

This example show how we can implement distributed queue using Redis and PHP.
It is very similar to the Distributed work pool / pipeline example, but instead of random work / messages processing, it use FIFO (first in - first out) functionality.

First file "master.php" that "push" messages.

For each message we do following:

Get next available ID using "INCR" command on "all:message" key.
Create a hash with parameters (hash fields "a")
Push the ID into a list called "queue:message".

Second file called "slave.php":

Pop next message from the list
Get the hash
Process the message

Because Redis is single threaded and atomic, we are insured that when some slave gets the message, no one slave will get same message.

Note that you can have many slaves and there is no requirement those slaves to be on same machine, so you can scale on as many machines as you want.

master.php

$r = new Redis();
$r->connect("127.0.0.1", "6379");


// Get new ID...
$message_id = $r->incr("all:message");

// Prepare the message, we use hash/array, but you can do it different way...
$message = array(
	"id"		=> $message_id		,
	"message_data"	=> rand()
);

// Set the hash...
$r->hmset("message:$message_id", $message);

// push the id into the queue...
$r->lpush("queue:message", $message_id);

slave.php

$r = new Redis();
$r->connect("192.168.0.1", "6379");

while(true){
	// pop the message ID from the queue, or block until one is available...
	$message_id = $r->blpop("queue:message");
	
	
	// Get the message itself...
	$message = $r->hgetall("message:$message_id");
	
	if (count($message) == 0)
		continue;
		
	// If you need, delete the key...
	$r->del("message:$message_id");
	
	// Process the message...
	process($message);
}

Redis connection note

The example uses generic connection to the Redis server. To make the example work with our service you will need to use code similar to this one:

// change these parameter according to the information in your instance list
$host = '50.30.35.9';
$port = 1234;
$password = 'somehashcode';
$db = 0;

$r = new Redis();
$r->connect($host, $port);
$r->auth($password);
$r->select($db);

blog comments powered by Disqus

Code library

Note:
Because we want this page to be useful for memcached users, we tagged with memcached all examples that may be "recreated" for memcached server.

Distributed queue example

master.php

slave.php

Redis connection note

Code library

MySQL cache in PHP

How to prevent website to be web-scraped

'Rotating' news

Show random element

Accounting / Vote example

Scoreboard example

Scaling example

Distributed work pool / pipeline example

Simple realtime web counter

Emulation of expiration of the set members

Lock example

Message queue example

Index example

Using Redis for MySQL autoincrement

Using Redis as "data buffer"

Calculating how many visitors are on the same webpage using rolling average