nexphant/queue

Job queue with workers, retry configurations, and graceful drain support.

Install

composer require nexphant/queue

Define a Job Handler

Job handlers extend Nexphant\Queue\JobHandler and implement the handle method:

use Nexphant\Queue\JobHandler;
use Nexphant\Queue\Job;

class SendEmailHandler extends JobHandler
{
    public function handle(array $payload, Job $job): void
    {
        $to = $payload['to'];
        $subject = $payload['subject'];
        
        // Process the background task
        mailer()->send($to, $subject);
    }
}

Setup and Register

Create a queue instance via QueueFactory and register the handler:

use Nexphant\Queue\QueueFactory;

$queue = QueueFactory::create([
    'driver' => 'database',
    'database' => $pdoConnection, // PDO connection instance
    'max_attempts' => 3,
    'retry_delay' => 60,
]);

// Register handler mapping a job name
$queue->register('send_email', SendEmailHandler::class);

Drivers

QueueFactory::create() can auto-select a driver, or you can choose one explicitly:

$memory = QueueFactory::create('memory');
$file = QueueFactory::create('file', ['path' => storage_path('queue')]);
$apcu = QueueFactory::create('apcu', ['prefix' => 'app_queue']);
$ring = QueueFactory::create('apcu_ring', ['scan_limit' => 2048]);
$redis = QueueFactory::create('redis', ['redis' => $redis]);
$database = QueueFactory::create('database', ['pdo' => $pdo]);

Driver options:

Driver	Required	Common Options
memory	none	In-process queue for tests and local jobs.
file	writable filesystem	path, max_payload_size, max_file_size.
database	pdo or database PDO instance	table, dead_letter_table.
apcu	ext-apcu	prefix, max_payload_size.
apcu_ring / lock_free	ext-apcu	prefix, max_payload_size, scan_limit.
redis	redis Redis instance	prefix, max_payload_size.

Dispatch Jobs

Push payload data using the registered job name:

// Dispatch immediately
$queue->push('send_email', [
    'to' => 'user@example.com',
    'subject' => 'Welcome to Nexphant',
]);

// Dispatch with a delay (in seconds)
$queue->later(60, 'send_email', [
    'to' => 'user@example.com',
    'subject' => 'Delayed Welcome',
]);

Run Workers

Start the worker process loop:

// Start worker loop listening on default queue
$queue->work();

Run a single job in tests or scripts:

$processed = $queue->workOnce(workerId: 1);

Worker and queue controls:

$status = $queue->status();
$metrics = $queue->metrics()->toArray();
$failed = $queue->failed(limit: 50);
$queue->retry($jobId);
$queue->stop();

QueueWorker wraps a long-running worker process with signal handling and limits:

use Nexphant\Queue\QueueWorker;

$worker = new QueueWorker($queue, [
    'max_jobs' => 1000,
    'max_time' => 3600,
    'memory_limit' => 256 * 1024 * 1024,
    'sleep' => 0.5,
]);

$worker->start();

Job Handler Lifecycle Hooks

The JobHandler abstract class provides lifecycle hooks that can be overridden:

class SendEmailHandler extends JobHandler
{
    public function handle(array $payload, Job $job): void { }

    public function before(Job $job): void
    {
        // Called before execution
    }

    public function after(Job $job, mixed $result): void
    {
        // Called after successful execution
    }

    public function failed(Job $job, \Throwable $e): void
    {
        // Called when execution fails
    }
}

Job Options

push() and later() accept per-job options:

Option	Description
max_attempts	Maximum retries before the job is moved to dead letters.
timeout	Job execution timeout in seconds.
metadata	Extra metadata stored with the job. Runtime context is added automatically when available.

Queue config also supports workers, retry_delay, timeout, memory_limit, poll_interval, metrics_interval, verbose, and quiet.

Related

• Queue Guide
• nexphant/core