Using MemcacheQ as message queue

Using a message queue can help to suspend heavy processes and execute them later so you won’t bother your visitors with long waiting times. There are a few solutions for queuing like Gearman, ActiveMQ and Zend Server Job Queue. For www.nd.nl (a Dutch newspaper) we wanted a simple and free queue mechanism that integrates with Zend Framework for handling a number of jobs. We found MemcacheQ.

MemcacheQ

MemcacheQ has some nice features that makes it an ideal candidate for our purpose:

• damn simple
• very fast
• multiple queues
• great concurrency
• memcache compatible protocol

Especially the last item is very interesting. This means that you can use the native memcache API for communicating with MemcacheQ, so you don’t need additional PHP extensions.

The installation has some dependencies like libevent and BerkleyDB. The latter is used for the persistent storage of the data inside the message queue. If something kills MemcacheQ you won’t loose the queued data.

MemcacheQ will run as a daemon and communicates over the UDP protocol. It can handle concurrent connections and can be used in a load-balanced environment.

Arguments

There are some arguments to use when starting the service. We’re currently using the following on www.nd.nl:

memcacheq -d -r -H /data/memcacheq -N -R -v -L 1024 -u nobody &> \
/data/memcacheq/error.log

Explanation of the used arguments:

• -d - Run as daemon
• -r - Maximize core file limit
• -v - Verbose logging
• -u <user> - Run as user nobody
• -H – Directory to store the BDB files
• -N – Performance improvement
• -R - Remove logfiles when no longer needed
• -L - Log buffer size, default is 32KB, we want 1024KB

Other helpful arguments:

• -h - Show the help with all the arguments
• -vv - More verbose mode
• Run without -d to directly see the output

Zend_Queue

Zend Framework provides an adapter to talk with MemcacheQ: Zend_Queue_Adapter_Memcacheq
This makes it easy to integrate the queue in your website.

Usage

We have a news site and want to show the number of visits to an article. When storing this number in a database, we need to execute a query every time a visitor reads an article. That’s something you don’t want because you can’t cache it and the visitor has to wait until the INSERT/UPDATE has finished. So we’ll have to store the hit temporarily in a fast queue and insert it later on in the database.

On the article page we add the new hit to the queue.

<?php
// MemcacheQ config
$queueOptions = array(
    'name' => 'example-queue',
    'driverOptions' => array(
        'host' => '127.0.0.1',
        'port' => 22201
    )
);

// Instantiate Zend_Queue
$queue = new Zend_Queue('MemcacheQ', $queueOptions);

// Find out if the queue exists
if (!$queue->getAdapter()->isExists($queueOptions['name']))
{
    // If not, create it
    $queue->createQueue($queueOptions['name']);
}

// Build a query string (key=val&key=val) because we need a scalar value
$params = http_build_query(
    array(
     'timestamp' => $timestamp,
     'page' => $page
    )
);
// Send the data to the queue
$queue->send($params);

And in a separate process (e.g. cronjob) we retrieve the queued hits and insert them in the database.

<?php
// MemcacheQ config
$queueOptions = array(
    'name' => 'example-queue',
    'driverOptions' => array(
        'host' => '127.0.0.1',
        'port' => 22201
    )
);

// Instantiate Zend_Queue
$queue = new Zend_Queue('MemcacheQ', $queueOptions);

// Retrieve 5 items from the queue
$messages = $queue->receive(5);

// $message is now a instance of Zend_Queue_Message_Iterator
// @TODO: Use a nice FilterIterator ;)
foreach ($messages as $job)
{
    if ('creating queue' == $job->body || false === $job->body)
    {
        // Skip message
        continue;
    }
    // Parse the query string to a array
    parse_str($job->body, $data);

    // Execute the heavy process
    $this->registerHit($data['timestamp'], $data['page']);
}

Known bugs

There’s currently (ZF version 1.11.2) one known bug in the ZF issue tracker: #ZF-9969.
This is a bug when using MemcacheQ 0.2.0.