Do you queue? Introduction to the Zend Server PHP Job Queue

There has been a lot of talk over the past several years about the difference between performance and scalability.  Never mind that the difference between the two will probably not really affect most developers.  Never mind that the “difference between performance and scalability” argument is often used when someone’s code performs poorly and their best argument is “Yeah, but my code scales”.  Yeah, sure it does.

But when talking about building a scalable application there is a big concept out there that many PHP developers are not overly familiar with.  That concept is queuing.  It is becoming much more prevalent in PHP-land but the concept of a queue is still relatively unused among PHP developers.

So, what is a queue?  Basically it means “take this and do something later”.  “This” could be anything, from a certain point of view (requisite Star Wars reference).  What that means is that “something” can be offloaded somewhere else (a queue) for further processing.  A queue is generally not an endpoint, but a conduit.  A pipe (requisite political reference).  But it is a pipe with a flow-control valve on it (requisite plumbing reference).  In other words the “something” will stay in the pipe until a) someone gets it, or b) it expires.  Hopefully, a.

This “something” is sometimes data and sometimes it is functionality.  There are a lot of data queues out there and the nice thing about data queues is that they are pretty much language independent.  In other words you can connect to a Java-based data queue from a PHP-based application and as long as you agree upon the format, like Stomp or JMS (if using the Java Bridge) then you can pass data back and forth without much problem.

However, there can be a problem when it comes to queueing functionality.  You clearly are not language independent.  Not that it’s a problem, but you’re not.  What this means is that now you have to have a specific method for implementing the queueing functionality.  There are a couple of open source options available, Gearman for one, but not many.  What I’d like to do is provide an example using the Job Queue in Zend Server 5.

Queueing a job is actually very easy to do.  A job is run by calling a URL where that job resides.  The Job Queue daemon will receive the request from your application and will then call the URL that you specified in the API call.  Once you call that URL your application can continue going on its merry way to finish serving up the request.

Serving the request

On your front end machine, the code to call the queue is pretty simple.  It consists of creating a ZendJobQueue object and calling the createHttpJob() method.  If you have any parameters that you need to pass to that job you can specify them in the second parameter of the call

1
2
3
4
5
6

$q->createHttpJob(
    'http://localhost/sendemail',
    array(
        'email' => $_POST['email']
    )
);

Then on the “sendemail” side your code would be

1
2
3
4
5

$params = ZendJobQueue::getCurrentJobParams();
if (isset($params['email'])) {
    mail($params['email'], 'Welcome', 'Welcome to my nightmare');
    ZendJobQueue::setCurrentJobStatus(ZendJobQueue::OK); 
}

That’s really all there is to it.

Or is there…

Serving the request… cool-y

My problem with this method is that it really is not as structured as I would like.  Modern applications are not really “scripts” even if they are written in a scripting language.  So, what I like doing is taking this existing functionality and providing some structure.  What I did for this website is take the existing Job Queue functionality and added something kinda similar to Java’s RMI.  It’s not quite, but kinda.  Or kinda like threading.  Not really, but kinda.

What I start out with is a generic abstract task class.  It looks like this.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

abstract class Esc_Queue_TaskAbstract
{
    const OPT_NAME = 'name';
    const OPT_SCHEDULE = 'schedule';
    const OPT_SCHEDULE_TIME = 'schedule_time';
 
    private $_options = array();
 
    protected abstract function _execute();
 
    public final function execute(Zend_Application $app, $qOptions = array())
    {
        $q = new ZendJobQueue();
        $jqOpts = $app->getOption('jobqueue');
        $qOptions = array_merge(
            array('name' => get_class($this)),
            $qOptions
        );
 
        $ret = $q->createHttpJob(
            $jqOpts['url'],
            array(
                'obj' => base64_encode(serialize($this))
            ),
            $qOptions
        );
        return $ret;
    }
 
    public final function run()
    {
        $this->_execute();
    }   
}

There are two defined methods and one abstract method.  The two defined methods are final because they need not and should not be overridden for the sake of predicability (final is under-used IMHO).  The execute() function doesn’t really execute anything.  It just takes the current class, serializes it and base64 encodes it, because the params don’t like binary data and sets it as a parameter called “obj”.  From there it inserts it into the Job Queue which is specified by a Zend_Application configuration setting.  That setting is

1

jobqueue.url = http://localhost/jq

Since queues generally contain privileged information it is a good idea to hide it from the outside world either on another machine/VM or web server directive.

The second method is called run().  It is not called on the front end machine.  The back end Job Queue will call that to execute the functionality that is defined in this class in the abstract method, called _execute().

So that’s the abstract class that our tasks are based off of, but how about an individual task?  What does that look like.  Well, to take our code that we had previously written…

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class Admin_Task_Mail extends Esc_Queue_TaskAbstract
{
 
    private $_email;
    private $_message;
    private $_subject;
 
    public function __construct($email, $subject, $message)
    {
        $this->_email     = $email;
        $this->_subject   = $subject;
        $this->_message   = $message;
    }
 
    public function _execute()
    {
        mail(
            $this->_email,
            $this->_subject,
            $this->_message
        );
    }
}

I put this code into my /application/modules/admin/tasks directory and added the following line to my bootstrap.

1

$al->addResourceType('task', 'tasks', 'Task');

That way the Zend_Application autoloader can easily autoload any tasks I have defined.

To execute this task, in my controllers, I simply type.

1
2
3
4
5
6
7
8
9

$mail = new Admin_Task_Mail(
    $_POST['to'],
    $_POST['subject'],
    $_POST['message']
);
 
$mail->execute(
    $this->getInvokeArg('bootstrap')->getApplication()
);

This will then send the job to the Job Queue daemon.

Speaking of.  We need to now execute our job.  That is done by defining a controller with code similar to the following.

1
2
3
4
5
6
7
8
9
10
11
12
13

$params = ZendJobQueue::getCurrentJobParams();
if (isset($params['obj'])) {
    $obj = unserialize(base64_decode($params['obj']));
    if ($obj instanceof Esc_Queue_TaskAbstract) {
        try {
            $obj->run();
            ZendJobQueue::setCurrentJobStatus(ZendJobQueue::OK);
            exit;
        } catch (Exception $e) {}
    }
}
ZendJobQueue::setCurrentJobStatus(ZendJobQueue::FAILED);
exit;

It retrieves the parameters and checks for one called “obj”.  It then unserializes the base64 decoded data, which should recreate the object that you created on the front end server.  After testing to make sure that it is an instance of Esc_Queue_TaskAbstract we call the run() method, which in turn calls the actual functionality we defined in _execute().

Sweet.

Summary

Key points on building super-cool job queue applications

1. Create an abstract class to wrap around your tasks
2. Use that abstract class to add itself to the Job Queue
3. Write a controller script that is the queue endpoint
4. Have that script recreate the object and execute the method you had defined in the code