Belajar Swoole - Kafka, RabbitMQ & Message Brokers

#Pendahuluan

Setelah di episode 13 kita mengelola koneksi database dengan pool — pada episode kali ini kita membahas pola arsitektur yang menjadi tulang punggung microservices: message brokers. Ketika sistem tumbuh, Anda tidak bisa lagi langsung memanggil service lain secara synchronous — Anda butuh antrian yang memisahkan producer dan consumer.

Mengapa episode ini penting? Karena Swoole adalah tempat yang sangat natural untuk menulis consumer: proses long-running dengan coroutine. Memakai PHP-FPM sebagai consumer berarti mengirim request HTTP setiap ada pesan — pemborosan besar. Dengan Swoole, consumer hidup terus dan memproses ribuan pesan per detik.

#Memilih Broker: Kafka, RabbitMQ, Redis Streams

Ketiganya dipakai lewat library klien PHP; yang membedakan adalah di mana kode tersebut dijalankan — dan di Swoole, jawabannya adalah proses long-running.

#Prinsip Coroutine Producer & Consumer

Model pesan asynchronous:

• Producer mengirim pesan — dari handler request, timer, atau task worker.
• Consumer membaca pesan terus-menerus dan memprosesnya.
• Coroutine memungkinkan satu consumer process menangani banyak pemrosesan sekaligus.

#Producer: Dari Handler Request

Mengirim pesan dari request adalah operasi cepat yang bisa ditembakkan langsung:

use Swoole\Coroutine;
 
Coroutine::create(function () {
    $conf = new RdKafka\Conf();
    $conf->set('bootstrap.servers', 'localhost:9092');
    $producer = new RdKafka\Producer($conf);
    $topic = $producer->newTopic('user-events');
 
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, json_encode([
        'type' => 'user_registered',
        'user_id' => 42,
        'at' => time(),
    ]));
    $producer->flush(5000);
});

Untuk RabbitMQ memakai php-amqplib, untuk Redis Streams cukup $redis->xadd('queue', '*', $data) — semua panggilan ini menjadi coroutine-friendly berkat hooking atau client coroutine Swoole.

#Consumer Worker dengan addProcess

Ini pola inti: sebuah proses custom yang berjalan terus di dalam server Swoole, membaca pesan tanpa henti:

<?php
use Swoole\Http\Server;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\Process;
 
$server = new Server('0.0.0.0', 9501);
$server->set(['worker_num' => 4]);
 
$consumer = new Process(function (Process $proc) use ($server) {
    // Koneksi konsumen Redis Streams
    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379);
 
    while (true) {
        $messages = $redis->xRead(
            ['user-events' => '>'], $count = 10, $block = 2000
        );
 
        if ($messages === false) {
            continue; // tidak ada pesan dalam 2 detik — cek lagi
        }
 
        foreach ($messages['user-events'] ?? [] as $id => $payload) {
            $server->sendMessage($payload, 0); // kirim ke worker untuk diproses
            $redis->xAck('user-events', 'swoole-group', [$id]);
        }
    }
}, false, SOCK_DGRAM);
 
$server->addProcess($consumer);
 
$server->on('Request', function (Request $req, Response $res) use ($redisPub) {
    $redisPub->xAdd('user-events', '*', ['event' => $req->post['event'] ?? 'ping']);
    $res->end('Pesan dikirim');
});
 
$server->on('PipeMessage', function ($server, $workerId, $message) {
    // Proses pesan di worker — coroutine context tersedia
    echo "Proses: " . json_encode($message) . "\n";
});
 
$server->start();

Kenapa pola ini kuat?

• Consumer dibesarkan di bawah Master — ikut di-manage (restart saat mati).
• Loop while (true) memakai xRead dengan block — tidak membakar CPU saat antrian kosong.
• sendMessage menyalurkan pesan ke worker yang punya coroutine context untuk pemrosesan.

#Acknowledgment dan Backpressure

Dua konsep yang menentukan kestabilan consumer:

Acknowledgment (ack): pesan yang berhasil diproses harus di-ack; yang gagal akan dikirim ulang oleh broker. Jangan ack sebelum pemrosesan benar-benar selesai — kalau crash di tengah, broker mengirim ulang dan pesan tidak hilang.

Backpressure: jika pemrosesan lebih lambat dari kedatangan pesan, antrian membengkak. Strateginya:

• Baca batch kecil ($count = 10) alih-alih borongan.
• Batasi concurrency pemrosesan dalam satu proses (channel + semaphore, episode 8).
• Naikkan jumlah consumer process bila perlu — di episode 22 kita skala ini.

#Common Pitfalls

#Penutup

Pada episode 14 ini, kalian telah mengintegrasikan Swoole dengan message brokers.

Inti yang harus dibawa pulang:

• Pilih broker sesuai workload: Kafka (streaming), RabbitMQ (routing kompleks), Redis Streams (ringan).
• Producer: kirim dari handler/timer; Consumer: proses long-running dengan coroutine.
• Consumer worker = addProcess + loop dengan blocking read + sendMessage ke worker.
• Acknowledge hanya setelah sukses; terapkan backpressure dengan batch kecil.

Di episode 15 selanjutnya, kita memakai semua fondasi ini lewat framework: Framework Integration — Hyperf, Swoft, dan Laravel Octane dengan driver Swoole. Sampai jumpa di episode 15!