Queue¶

Note

Not part of core. Install it separately:

composer require kinetis/queue

A backend-agnostic background job queue: push a job from application code, a separate vendor/bin/queue worker process pops and runs it. Redis and SQL (MySQL/Postgres) backends are both included, behind one Kinetis\Queue\QueueInterface contract — application code never sees which one is running underneath. Amazon SQS (Queue (SQS)) and RabbitMQ (Queue (RabbitMQ)) are two more, equally QueueInterface-conforming backends, each in its own separate package.

Writing a job¶

A job is a plain class implementing Kinetis\Queue\Job, constructed the same way as any other DTO in Kinetis:

use Kinetis\Queue\Job;

final readonly class SendWelcomeEmail implements Job
{
    public function __construct(
        public string $email,
        public string $name,
    ) {}

    public function handle(Mailer $mailer): void
    {
        $mailer->send($this->email, "Welcome, {$this->name}!");
    }
}

handle()’s own parameters are resolved through the container at run time, the same way a controller method’s parameters are — not the constructor. A job’s constructor only ever holds the data it needs to survive being written to the queue and read back later by a worker process, which may not be the same process (or even the same machine) that pushed it.

Pushing and processing¶

use Kinetis\Queue\QueueInterface;

final readonly class RegistrationController
{
    public function __construct(private QueueInterface $queue) {}

    #[Post('/register')]
    public function store(#[Body] RegisterRequest $data): array
    {
        // ...
        $this->queue->push(new SendWelcomeEmail($data->email, $data->name));

        return ['status' => 'registered'];
    }
}

vendor/bin/queue work

The worker runs forever, one job at a time: pop, resolve handle()’s dependencies through a fresh container scope, invoke it, then pop the next one.

Named and prioritized queues¶

Every push() targets a named queue — default when none is given:

$this->queue->push(new SendWelcomeEmail($data->email, $data->name));
$this->queue->push(new GenerateReport($reportId), queue: 'reports');
$this->queue->push(new SendPasswordReset($data->email), queue: 'high');

A worker watches one or more queue names, in priority order:

vendor/bin/queue work --queue=high,default

Priority is expressed by list order, not a numeric per-job score: the worker above drains everything on high before ever checking default. A queue name absent from --queue is invisible to that worker entirely — jobs pushed to reports sit untouched until some worker actually watches reports. Omitting --queue watches default only.

Choosing a backend¶

QUEUE_CONNECTION has no default — set it explicitly to redis, sql, sqs, or rabbitmq:

QUEUE_CONNECTION=redis
REDIS_HOST=127.0.0.1

# — or —

QUEUE_CONNECTION=sql
DB_CONNECTION=mysql   # or "pgsql"
DB_HOST=127.0.0.1
DB_NAME=app
DB_USER=app
DB_PASSWORD=secret

# — or —

QUEUE_CONNECTION=sqs
QUEUE_SQS_REGION=us-east-1

# — or —

QUEUE_CONNECTION=rabbitmq
QUEUE_RABBITMQ_URL=amqp://guest:guest@localhost:5672/

The Redis variables are the same ones Persistence’s RedisSimpleCache reads; the SQL variables are the same ones Migrations reads. sqs needs the separate kinetis/queue-sqs installed — see Queue (SQS) for its own configuration. rabbitmq needs kinetis/queue-rabbitmq — see Queue (RabbitMQ).

QUEUE_CONNECTION_NAME picks which named connection of that backend this worker uses (see Configuration’s named-connection convention) — ‘default’ when unset:

QUEUE_CONNECTION=redis
QUEUE_CONNECTION_NAME=cache2
REDIS_CACHE2_HOST=127.0.0.1

Scaling out: multiple workers¶

vendor/bin/queue work is safe to run as any number of separate, concurrent processes against the same backend — a plain horizontal-scaling lever, not something that needs coordinating by hand. Every backend guarantees a job is handed to exactly one worker, so two workers running at once never both pick up the same job. Workers don’t need to agree on anything beyond which queue names they watch — start as many as you want, on as many machines as you want, pointed at the same backend.

Multiple backends¶

Different queues can live on different backends — a RedisQueue for low-latency jobs, a SqlQueue for jobs that should ride along with an existing database’s backups and transactions, an SqsQueue (from Queue (SQS)) for jobs a separate AWS account or service needs to consume, or a RabbitMqQueue (from Queue (RabbitMQ)) for jobs routed through an existing broker. Register each concrete class directly rather than binding QueueInterface to just one of them:

$app->instance(RedisQueue::class, new RedisQueue($redisClient));
$app->instance(SqlQueue::class, new SqlQueue($db));

Application code that pushes to a specific backend constructor-injects that concrete class instead of the shared interface:

final readonly class RegistrationController
{
    public function __construct(private RedisQueue $fastQueue) {}
}

Each backend gets its own vendor/bin/queue work process, run with the QUEUE_CONNECTION/backend-specific environment variables and --queue flag matching the queue names pushed to it — one process per backend-and-queue-set combination, same as scaling a single backend out.

Running synchronously in development: `SyncQueue`¶

use Kinetis\Queue\SyncQueue;

if ($appEnv->isDevelopment()) {
    $app->instance(QueueInterface::class, new SyncQueue($app));
}

push() runs the job’s handle() immediately, inline, rather than storing it anywhere — no separate vendor/bin/queue work process needed while developing locally. Every push() call still gets its own fresh scope, exactly like a real worker gives each job — job code that happens to depend on request-scoped state reachable only by accident would fail the same way here as it would against a real backend, rather than silently working in development and breaking once actually queued. Unlike a real worker, a failing job’s exception isn’t caught — it propagates straight to whatever called push(), so the point of running jobs synchronously (seeing the real error immediately) still holds.

Not selectable via QUEUE_CONNECTION — there’s nothing for a worker process to do against a backend that never stores anything, so SyncQueue is constructed directly in application bootstrap code instead of through bin/queue. It accepts a queue argument on push() for signature compatibility with the other backends, but ignores it — there’s only ever one “queue” (immediate execution), so a queue name has nothing to select between.

The SQL backend needs a table¶

kinetis/queue ships two ready-to-copy Migrations files — one per dialect, since the auto-incrementing primary key syntax itself isn’t portable between MySQL and Postgres:

vendor/kinetis/queue/resources/migrations/create_kinetis_queue_jobs_table.mysql.php.stub
vendor/kinetis/queue/resources/migrations/create_kinetis_queue_jobs_table.pgsql.php.stub

Copy whichever matches your database into your own migrations/ directory with a timestamp prefix, then run vendor/bin/migrate migrate.

SqlQueue’s pop() relies on SELECT ... FOR UPDATE SKIP LOCKED to guarantee two workers never receive the same job — that’s this backend’s actual version floor: MySQL 8.0+ or MariaDB 10.6+. An older server doesn’t support that clause at all, so pop() would fail outright rather than degrade quietly.

Delayed jobs¶

$this->queue->push(new SendReminderEmail($userId), delaySeconds: 3600);

A delayed job isn’t visible to pop() until its delay has elapsed. The Redis and SQL backends check this on their own polling cycle rather than firing at the exact moment the delay ends, so a delayed job can run slightly later than its exact target time — typically by a few seconds, not less. SQS’s own delay is native (SendMessage’s own DelaySeconds, no polling-based promotion at all) but capped at 900 seconds (15 minutes) — see Queue (SQS). RabbitMQ’s delay is also broker-driven, with no such cap — see Queue (RabbitMQ).

A failing job¶

A job whose handle() throws is logged through whatever Psr\Log\LoggerInterface is registered — it doesn’t stop the worker, which moves on to the next job either way. What happens to the job itself depends on maxAttempts:

$this->queue->push(new SendWelcomeEmail($data->email, $data->name), maxAttempts: 3);

maxAttempts is set per job at push() time, and takes priority whenever it’s set. Once a job’s attempt count reaches it, that attempt’s failure is final: the job is removed instead of being retried again.

A job pushed without maxAttempts falls back to vendor/bin/queue work’s own default instead, QUEUE_MAX_ATTEMPTS in .env — 0 when unset, meaning no retries at all: a job that fails once is given up on immediately.

QUEUE_MAX_ATTEMPTS=5

Raise it to allow retries. There is no configuration, anywhere, that makes a job retry forever — QueueWorker’s own default is likewise 0, not unlimited, so a job with no maxAttempts of its own is only ever retried if something has explicitly set a cap above 1.

{
    "level": "error",
    "message": "Job \"App\\SendWelcomeEmail\" failed permanently after 3 attempt(s): Connection refused",
    "context": {
        "job": {"class": "App\\SendWelcomeEmail", "args": {"email": "a@b.com", "name": "Ana"}},
        "exception": "..."
    }
}

That log entry — job class, constructor arguments, and the exception — is the only record kept of a job that’s given up on; there’s no dead-letter table or queue to inspect afterward. A job that still has attempts remaining logs the same way, without “permanently” in the message, and is released rather than removed.

Deferring an event listener onto the queue¶

Kinetis\Queue\QueuedListenerInvoker implements core’s Kinetis\Events\ListenerInvokerInterface — a listener marked Kinetis\Events\ShouldQueue runs as a real queued job instead of inline:

use Kinetis\Events\ListenerInvokerInterface;
use Kinetis\Queue\QueuedListenerInvoker;

$app->instance(ListenerInvokerInterface::class, new QueuedListenerInvoker($queue));

See Events for writing the listener itself.

Queue¶

Writing a job¶

Pushing and processing¶

Named and prioritized queues¶

Choosing a backend¶

Scaling out: multiple workers¶

Multiple backends¶

Running synchronously in development: SyncQueue¶

The SQL backend needs a table¶

Delayed jobs¶

A failing job¶

Deferring an event listener onto the queue¶

See also¶

Running synchronously in development: `SyncQueue`¶