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.