2-queue-manager.md

The QueueManager

The WorkerBundle creates a QueueManager service which acts as a central point to add jobs in the queue, receive them back from it, inspect, bury and delete jobs.

Examples

Below are some basic usage examples. These examples use a tube (the name of a queue in Beanstalk parlance) named example.

Add a job:

$queueManager->add('example', $payload = ['data']);

Receive a job:

# watch the 'example' tube
$queueManager->watch('example');

# waits at most 10 seconds, returns a \Pheanstalk\Job instance
$queueManager->get(10);

Inspect a job. This returns the next-ready job from the queue, without actually reserving the job from the queue:

# returns a \Pheanstalk\Job instance
$queueManager->peek('example');

Bury a job. This 'parks' the job outside the queue, which means the job will not be released until it's kicked back onto the queue:

$queueManager->bury($job);

Kick jobs back onto the queue:

$queueManager->kick('example', $max = 10);

Delete a job:

$queueManager->delete($job);

Clear a tube:

$queueManager->clear('example');

Adding for objects

Often the jobs for a tube correspond directly to an object; a model in the application's domain. For example: a user.mail.registration tube has a direct relation to a User object. In these cases it can be useful to pass objects to the QueueManager instead of constructing the payload manually every time. We'll explain how these payloads are constructed in the next chapter, for now we'll show how you this example works:

$queueManager->addForObject('user.mail.registration', $user);

Time to run

When adding a job you can indicate how long a worker may take to process this job, before the job is considered released. This is called the time to run. Since the queue is asynchronous it does not get an immediate response from the worker after reserving a job for it. After the time to run has expired, and the job has not been deleted, the queue considers the job as gone wrong and releases it, after which the job becomes available to reserve again.

By default the time to run is set to 2 minutes. If you have a job that takes a long time to complete, make sure to increase the time to run, or to touch the job during the process.

Delaying & prioritizing jobs

The add() and addForObject() methods accept more arguments to give you more control over when the job will be reserved to a worker.

You can set a specific time after which the job can be reserved:

$queueManager->addForObject('user.mail.registration', $user, '10 minutes');

This delays the job for 10 minutes after now.

The second argument to influence this is the job priority. The priority is a number between 0 (most urgent) and 4294967295 (least urgent). By default, all jobs get a default priority of 1024, but you can change this in the add* method calls:

# urgent
$queueManager->addForObject('user.mail.registration', $user, null, 10);

# low priority
$queueManager->addForObject('user.mail.registration', $user, null, 10000);

Combining these arguments, you can influence even more control here:

# urgent, but only after 1 minute
$queueManager->addForObject('user.mail.registration', $user, '1 minute', 10);

Note how we're saying influence here. Depending on the number of jobs in the queue, and the number of workers that are available, a job may come available after the designated time, or not immediately if you're using the highest priority. If there are 1000 jobs scheduled for right now, and there are 10 available workers, it may take some time for the queue to be completely processed.

Next

Now that we know how to manage jobs, let's see how we can actually work them: Executors.