Queues And Jobs
Some work should not happen during a web request. Sending a welcome email, resizing an uploaded image, generating a PDF report — these tasks take time, and making a user wait for them is a poor experience. Queues solve this by deferring work to a background process that runs independently of the HTTP cycle.
In PHP, a queue is a data structure that holds pending jobs. A worker process pulls jobs off the queue one by one and executes them. The web request adds a job and returns immediately; the worker handles the rest.
This lesson builds a complete queue system from scratch using only PHP. No Redis, no database — just the core patterns you will find in every production queue library.
The Job Interface
Every item in a queue is a job — a self-contained unit of work. The cleanest way to model this is with an interface. Every job must know how to execute itself:
<?php
interface Job {
public function handle(): void;
public function getName(): string;
}
class SendWelcomeEmail implements Job {
public function __construct(private string $email) {}
public function handle(): void {
// In production this would call a mailer
echo "Sending welcome email to: {$this->email}\n";
}
public function getName(): string {
return 'SendWelcomeEmail';
}
}
class GenerateReport implements Job {
public function __construct(private string $reportType) {}
public function handle(): void {
echo "Generating {$this->reportType} report...\n";
}
public function getName(): string {
return 'GenerateReport';
}
}
$job = new SendWelcomeEmail('alice@example.com');
$job->handle();
$report = new GenerateReport('monthly-sales');
$report->handle();
The interface enforces a contract: any class that implements Job can be dispatched to the queue without the queue needing to know anything about what the job actually does. This is the open/closed principle in practice.
Building a Simple Queue
PHP's SplQueue is a doubly-linked list that operates in FIFO order — first in, first out. Jobs are pushed onto the back and pulled from the front:
<?php
interface Job {
public function handle(): void;
public function getName(): string;
}
class Queue {
private SplQueue $jobs;
public function __construct() {
$this->jobs = new SplQueue();
$this->jobs->setIteratorMode(SplDoublyLinkedList::IT_MODE_FIFO);
}
public function push(Job $job): void {
$this->jobs->enqueue($job);
echo "Queued: {$job->getName()}\n";
}
public function pop(): ?Job {
if ($this->jobs->isEmpty()) {
return null;
}
return $this->jobs->dequeue();
}
public function size(): int {
return $this->jobs->count();
}
}
class ResizeImage implements Job {
public function __construct(private string $path, private int $width) {}
public function handle(): void {
echo "Resizing {$this->path} to {$this->width}px wide\n";
}
public function getName(): string { return 'ResizeImage'; }
}
class NotifyAdmin implements Job {
public function handle(): void {
echo "Notifying admin of new signup\n";
}
public function getName(): string { return 'NotifyAdmin'; }
}
$queue = new Queue();
$queue->push(new ResizeImage('uploads/photo.jpg', 800));
$queue->push(new NotifyAdmin());
$queue->push(new ResizeImage('uploads/avatar.png', 200));
echo "\nQueue size: " . $queue->size() . "\n\n";
while ($job = $queue->pop()) {
echo "Processing: {$job->getName()}\n";
$job->handle();
echo "---\n";
}
Notice how the queue does not import ResizeImage or NotifyAdmin directly — it only knows about the Job interface. Adding a new job type requires zero changes to the queue itself.
Adding Retry Logic
Real jobs fail. A network timeout, a file permission error, a third-party API going down. A robust queue retries failed jobs up to a maximum attempt count before giving up:
<?php
interface Job {
public function handle(): void;
public function getName(): string;
public function getMaxAttempts(): int;
}
class JobWrapper {
public int $attempts = 0;
public function __construct(public readonly Job $job) {}
}
class RetryQueue {
private array $pending = [];
private array $failed = [];
public function push(Job $job): void {
$this->pending[] = new JobWrapper($job);
}
public function process(): void {
while (!empty($this->pending)) {
$wrapper = array_shift($this->pending);
$wrapper->attempts++;
try {
echo "Attempt {$wrapper->attempts} for: {$wrapper->job->getName()}\n";
$wrapper->job->handle();
echo " OK\n";
} catch (RuntimeException $e) {
echo " FAILED: {$e->getMessage()}\n";
if ($wrapper->attempts < $wrapper->job->getMaxAttempts()) {
$this->pending[] = $wrapper; // re-queue
} else {
$this->failed[] = $wrapper;
echo " Moved to failed queue after {$wrapper->attempts} attempts\n";
}
}
}
if (!empty($this->failed)) {
echo "\nFailed jobs: " . count($this->failed) . "\n";
foreach ($this->failed as $w) {
echo " - {$w->job->getName()}\n";
}
}
}
}
class FlakyApiCall implements Job {
private static int $callCount = 0;
public function handle(): void {
self::$callCount++;
if (self::$callCount < 3) {
throw new RuntimeException('API timeout');
}
echo " API call succeeded\n";
}
public function getName(): string { return 'FlakyApiCall'; }
public function getMaxAttempts(): int { return 3; }
}
class AlwaysFails implements Job {
public function handle(): void {
throw new RuntimeException('Service unavailable');
}
public function getName(): string { return 'AlwaysFails'; }
public function getMaxAttempts(): int { return 2; }
}
$queue = new RetryQueue();
$queue->push(new FlakyApiCall());
$queue->push(new AlwaysFails());
$queue->process();
The JobWrapper tracks attempt count separately from the job itself. The job stays clean and focused on business logic; the infrastructure handles retry bookkeeping.
Priority Queues
Not all jobs are equally urgent. A password reset email should go out before a weekly digest. PHP's SplPriorityQueue handles this — lower priority numbers mean higher urgency (or you can invert this convention):
<?php
interface Job {
public function handle(): void;
public function getName(): string;
public function getPriority(): int;
}
class PriorityQueue {
private SplPriorityQueue $queue;
public function __construct() {
$this->queue = new SplPriorityQueue();
}
public function push(Job $job): void {
// Higher priority number = processed first in SplPriorityQueue
$this->queue->insert($job, -$job->getPriority());
}
public function process(): void {
while (!$this->queue->isEmpty()) {
$job = $this->queue->extract();
echo "Running [{$job->getPriority()}] {$job->getName()}\n";
$job->handle();
}
}
}
class PasswordResetEmail implements Job {
public function __construct(private string $email) {}
public function handle(): void { echo " Password reset sent to {$this->email}\n"; }
public function getName(): string { return 'PasswordResetEmail'; }
public function getPriority(): int { return 1; } // highest urgency
}
class WeeklyDigest implements Job {
public function handle(): void { echo " Weekly digest compiled\n"; }
public function getName(): string { return 'WeeklyDigest'; }
public function getPriority(): int { return 10; } // lowest urgency
}
class NewUserNotification implements Job {
public function handle(): void { echo " New user notification sent\n"; }
public function getName(): string { return 'NewUserNotification'; }
public function getPriority(): int { return 5; }
}
$queue = new PriorityQueue();
// Push in arbitrary order — priority determines execution order
$queue->push(new WeeklyDigest());
$queue->push(new PasswordResetEmail('bob@example.com'));
$queue->push(new NewUserNotification());
$queue->push(new PasswordResetEmail('carol@example.com'));
$queue->process();
The queue processes jobs by urgency regardless of insertion order. The password resets go first, then the notification, and the digest runs last.
Try It Yourself
Build a job queue that supports delayed execution. Each job has an optional delay in seconds. Jobs whose delay has not yet elapsed should be skipped and re-checked on the next pass:
<?php
interface Job {
public function handle(): void;
public function getName(): string;
}
class DelayedQueue {
private array $items = [];
public function push(Job $job, int $delaySeconds = 0): void {
$this->items[] = [
'job' => $job,
'runAt' => time() + $delaySeconds,
];
echo "Queued: {$job->getName()} (delay: {$delaySeconds}s)\n";
}
public function tick(): void {
$now = time();
$remaining = [];
foreach ($this->items as $item) {
if ($item['runAt'] <= $now) {
echo "Executing: {$item['job']->getName()}\n";
$item['job']->handle();
} else {
$remaining[] = $item;
}
}
$this->items = $remaining;
echo "Pending jobs: " . count($this->items) . "\n";
}
}
class BackupDatabase implements Job {
public function handle(): void { echo " Database backup complete\n"; }
public function getName(): string { return 'BackupDatabase'; }
}
class CleanTempFiles implements Job {
public function handle(): void { echo " Temp files cleaned\n"; }
public function getName(): string { return 'CleanTempFiles'; }
}
class SendNewsletter implements Job {
public function handle(): void { echo " Newsletter dispatched\n"; }
public function getName(): string { return 'SendNewsletter'; }
}
$queue = new DelayedQueue();
$queue->push(new BackupDatabase(), 0); // run immediately
$queue->push(new CleanTempFiles(), 0); // run immediately
$queue->push(new SendNewsletter(), 60); // would wait 60 seconds in production
echo "\n--- Tick ---\n";
$queue->tick();
Try adding a getDelay() method to the Job interface so jobs carry their own default delay, or extend tick() to accept a simulated timestamp for testing without sleep().
Key Takeaways
- A job is a self-contained unit of work defined by an interface — the queue only cares about the contract, not the implementation.
- SplQueue provides FIFO ordering; SplPriorityQueue adds urgency-based scheduling using a numeric priority.
- Retry logic belongs in the queue infrastructure, not in the job itself — keep jobs focused on business logic.
- A delayed queue defers execution until a future timestamp, enabling scheduled background processing.
- In production, the queue backend (Redis, database, SQS) stores serialized jobs so workers can run in separate processes; the patterns here are identical.
- Idempotency is critical — design jobs so that running them twice produces the same result, because retries are inevitable.
Pro Tip: When serializing jobs for a real queue backend, store only the minimal data needed — IDs rather than full objects. Load the full record inside
handle(). This prevents stale data bugs where the job was queued with one version of a record and the record changed before the job ran.
Next Steps
You can now implement job queues with FIFO ordering, priority scheduling, retry logic, and delayed execution. Your applications can offload heavy work to the background.
Background jobs solve the problem of slow operations, but what about operations that are fast individually yet run thousands of times? Querying the same database row on every request, recalculating the same result for every visitor, fetching the same API response repeatedly -- these add up. Caching and performance teaches you to store computed results so your application avoids redundant work, responds faster, and handles more traffic with the same resources.