Belajar RoadRunner - Queue & Jobs Plugin

#Pendahuluan

Setelah di episode 10 kita menyimpan data dengan KV, pada episode kali ini kita memindahkan pekerjaan ke background: plugin queue & jobs RoadRunner. Email notifikasi, laporan PDF, sinkronisasi data, webhook — semua tugas yang tidak boleh memperlambat respons HTTP — kita kirim ke antrian dan diproses oleh worker khusus.

Mengapa episode ini penting? Karena inilah cara aplikasi PHP modern memisahkan fast path (merespons user) dari slow path (kerja berat). Dan karena RoadRunner mengelola konsumsi antrian di level server Go, kalian tidak perlu daemon queue terpisah seperti yang biasa dilakukan Laravel/Beanstalkd.

#Arsitektur Queue & Jobs

Dua konsep yang perlu dibedakan:

100%

Pipeline: koneksi ke antrian aktual (driver + nama queue).
Job: unit kerja yang dikirim (payload + option seperti delay, priority).
Consumer: worker khusus yang menarik job dari antrian dan mengeksekusinya.

Kelebihan RoadRunner: satu binary mengelola HTTP dan jobs — tidak ada supervisor terpisah untuk daemon queue.

#Konfigurasi Pipeline

Plugin jobs mendukung banyak driver: redis, amqp, sqs, beanstalk, dan boltdb. Contoh tiga pipeline:

Blok jobs dengan beberapa pipeline

jobs:
  pool:
    num_workers: 4
    max_worker_memory: 128
  consume:
    - queue: "emails"
      pipeline: "redis-emails"
    - queue: "reports"
      pipeline: "amqp-reports"
  pipelines:
    redis-emails:
      driver: redis
      config:
        queue: emails
        prefetch: 10
    amqp-reports:
      driver: amqp
      config:
        addr: amqp://guest:guest@localhost:5672/
        queue: reports
        exchange: reports
    sqs-uploads:
      driver: sqs
      config:
        key: ""
        secret: ""
        endpoint: https://sqs.us-east-1.amazonaws.com
        queue: uploads

consume memberi tahu server queue mana yang harus dikonsumsi dan oleh pipeline mana. Setiap pipeline bisa memakai pool berbeda — beban email tidak boleh mengganggu pemrosesan laporan.

#Mendorong Job dari Aplikasi

Dari sisi PHP, pakai package spiral/roadrunner-jobs:

Install package jobs

composer require spiral/roadrunner-jobs

Push job ke antrian

use Spiral\RoadRunner\Jobs\Jobs;
 
$jobs = new Jobs($rpc);
 
// Pilih pipeline
$queue = $jobs->connect('redis-emails');
 
// Buat task dengan payload
$task = $queue->create(
    'email.send',
    ['to' => 'user@example.com', 'template' => 'welcome'],
);
 
// Option opsional
$task->withDelay(5);          // tunda 5 detik
$task->withPriority(2);       // prioritas
 
// Kirim
$queue->push($task);

connect('redis-emails') memakai nama pipeline dari config. create() menerima nama job (bebas) dan payload. push() mengirimnya ke antrian via RPC — eksekusi dilakukan pipeline, bukan oleh worker HTTP.

#Menulis Consumer Worker

Consumer adalah worker PHP biasa — file consumer.php yang dijalankan pipeline. Pakai package spiral/roadrunner-jobs sisi consumer:

consumer.php - proses job

<?php
 
declare(strict_types=1);
 
use Spiral\RoadRunner\Jobs\Consumer;
use Spiral\RoadRunner\Jobs\Serializer\JsonSerializer;
use Spiral\RoadRunner\Worker;
 
require __DIR__ . '/vendor/autoload.php';
 
$consumer = new Consumer(Worker::create(), new JsonSerializer());
 
while ($task = $consumer->waitTask()) {
    try {
        $payload = $task->getPayload();
 
        if ($payload['template'] === 'welcome') {
            sendWelcomeEmail($payload['to']);
        }
 
        $task->ack();
    } catch (\Throwable $e) {
        $task->error('Gagal memproses: ' . $e->getMessage());
    }
}

Siklus hidup task penting:

Aksi	Arti
`ack()`	Job selesai sukses — dihapus dari antrian
`error()`	Job gagal — dicatat, masuk statistik, tidak di-redeliver
`fail()`	Job gagal dan akan di-redeliver (retry)
`requeue()`	Kembalikan ke antrian dengan delay tertentu
(tidak dipanggil)	Worker mati sebelum menjawab — job di-redeliver otomatis

Tip

Untuk job yang boleh retry dengan backoff, gunakan $task->requeue($delay). Untuk job yang tidak boleh retry (misal payload invalid), gunakan error() — jangan buang resource dengan retry tanpa batas.

#IDEMPOTENSI: Aturan Emas Job

Karena job bisa di-redeliver (misal worker mati setelah memproses tapi sebelum ack), setiap job harus idempoten — mengeksekusinya dua kali tidak boleh merusak apa pun. Contoh: sebelum mengirim email, cek di KV store apakah email:sent:{task_id} sudah ada:

Pola idempotent

$taskId = $task->getHeader('id') ?? $task->getId();
$doneKey = 'job:done:' . $taskId;
 
if ($cache->get($doneKey)) {
    $task->ack();
    continue;
}
 
sendEmail($payload['to']);
$cache->set($doneKey, true, 3600);
$task->ack();

#Error & Monitoring Queue

Gejala umum dan solusinya:

Gejala	Penyebab	Solusi
Job tidak pernah diproses	Pipeline tidak terdaftar di `consume`	Pastikan pipeline ada di daftar `consume`
`connection refused` Redis	Redis tidak jalan	Cek Redis dan `addr` pipeline
Worker jobs macet	Consumer tidak memanggil ack/error	Selalu akhiri task dengan salah satu
Job duplikat	Retry tanpa idempotensi	Terapkan pola idempotent di atas

Statistik antrian bisa dilihat lewat ./rr jobs:list dan detail per job — kita kombinasikan dengan metrics di episode 12.

#Penutup

Pada episode 11 ini, kalian telah menguasai queue & jobs.

Inti yang harus dibawa pulang:

Pipeline = koneksi ke antrian (Redis, AMQP, SQS, Beanstalk, Boltdb).
Push job dari aplikasi dengan spiral/roadrunner-jobs via RPC.
Consumer worker menarik task, memproses, lalu ack/error/requeue.
Job harus idempoten — redelivery adalah hal normal.
Satu binary mengelola HTTP dan jobs; consume menentukan alur kerja.

Di episode 12 selanjutnya, kita belajar metrics & Prometheus — membuka endpoint /metrics, mendefinisikan counter, gauge, dan histogram kustom, lalu memvisualisasikan latensi request dan kesehatan pool di dashboard. Sampai jumpa di episode 12!