chh / kue

Generic interface to job queues for PHP projects.

Kue

A minimalistic, generic and framework independent interface to job queues

Install

Install via Composer:

% wget http://getcomposer.org/composer.phar
% php composer.phar require chh/kue:*@dev

Use

Jobs

Jobs are classes which implement the Kue\Job interface. This interface specifies one method — run(). The run method is invoked by the worker script, which pulls jobs out of the queue.

---

Why not just implement callable?

Because if the contract is that a job must be callable, then every callback (even a Closure) can satisfy it. The problem with this approach is, that workers usually run in an other process than the script that puts the jobs into the queue. Closures and functions can't be serialized, so there's no way to transport them to the worker process. Objects can be serialized in one process and unserialized in another, and retain all their state and context.

---

Queues

A queue implements the Kue\Queue interface. This interface looks as follows:

interface Kue\Queue
{
    function push(Kue\Job $job);
    function pop();
    function flush();
    function process(Kue\Worker $worker);
}

The queue's responsibility is, to be the transport layer between the user who pushes jobs into the queue, and the worker script which pulls them out by polling pop() for jobs to process.

The flush() method should be called by the client after jobs have been pushed, and can be used to send multiple jobs with a more efficient transport mechanism — like a batch request.

Kue ships with a Kue\LocalQueue for development environments. This queue sends jobs to a local network socket on push(), and receives on this network socket when pop() is called.

For production it's recommended to use something like SQS behind the Queue interface.

Workers

Workers take the queue, and start polling with the pop() method for jobs.

interface Kue\Worker extends \Evenement\EventEmitterInterface
{
    function process(Kue\Queue $queue);
}

Workers should abstract the strategy for processing the queued jobs. The worker implementations which are shipped with Kue should be fully sufficient for most use cases.

Kue ships with these workers out of the box:

• Kue\SequentialWorker — a simple worker which calls the queue's pop() method in a for loop, and processes jobs strictly one after the other. This is simpler and works on any platform, but should be only used for development or on Windows.
• Kue\PreforkingWorker — an advanced worker, which starts a master process and forks a pool of worker processes, which all call pop() individually. It can run jobs concurrently and is more fault tolerant, because when a worker dies, the master simply starts up a new one. You will want to use this worker in production. It only works on *nix platforms, like OSX and Linux.

All workers are Evenement Event Emitters and emit following standard events:

• init: Emitted before the job gets run, gets called with the job as argument.
• success: A job was processed successfully. The handler receives the job as argument.
• exception: An exception occured while running the job. The job and exception are passed as arguments.

The Kue\PreforkingWorker supports following additional events:

• beforeFork: This is emitted before the worker process gets forked from the parent.
• afterFork: Emitted from the child process after it was forked from the master. Use this to reinitialize resources in the worker process.

These workers are typically managed by the Symfony Console Command, which ships with Kue:

use Symfony\Component\Console\Application;
use Kue\Command\WorkCommand;
use Kue\LocalQueue;

$worker = new WorkCommand(new LocalQueue);

$app = new Application;
$app->add($worker);

$app->run();

The queue worker can the be run with the kue:work command inside the Symfony Console Application. The worker implementations can be switched with the -c flag:

• 1: The Kue\SequentialWorker is used.
• >1: The Kue\PreforkingWorker is used, with a pool of c workers.

To attach your own event listeners to the automatically selected worker instance, you can pass an array of event names plus handlers as second argument to the command's constructor:

$webApp = new MyApplication;

$worker = new WorkerCommand(new LocalQueue, array(
    'init' => function($job) use ($webApp) {
        $job->application = $webApp;
    }
));

kue's Issues

Run job's scheduled by Cron expressions only once within a minute

Add ImmediateQueue as a helper for automated tests

Add an ImmediateQueue queue implementation which calls the job's run method immediately after the job was pushed. This is useful for tests.

Add support for scheduled jobs

Add a separate kue:scheduler command which takes a queue and puts a set of jobs at configured intervals into the queue.

• Use Cron syntax to define intervals or a DSL?
• Sleep until the next job is ready to be scheduled
• Add a class Kue\Scheduler, and a kue:scheduler command which provides a Symfony Console Command and uses the scheduler.

Scheduler API:

$scheduler = new Kue\Scheduler(new Kue\LocalQueue);

$scheduler->every("5m", new HelloWorldJob);

for (;;) {
    # Returns after a job was scheduled, or when the max wait time was reached
    $scheduler->run();
}

Command usage:

$scheduler = new Kue\Scheduler(new Kue\LocalQueue);
$scheduler->every("15m", new ExpensiveStuffJob);

$app = new Symfony\Component\Console\Application;
$app->add(new Kue\Command\Scheduler($scheduler));
$app->run();

Redis errors when using PreForkingWorker

When using redis backed queue with multiple workers, none of the forks can reach redis so it ends up dumping "read error on connection"

A solutions for this could be to pass in a factory or closure to create the RedisQueue

Add Amazon SQS Queue