Pub/Sub dengan Redis — Realtime

Redis Pub/Sub Saat real-time app berjalan di multiple server instances, event yang terjadi di server A harus sampai ke client yang terkoneksi di server B…

Pub/Sub (publish/subscribe) adalah pola kirim pesan: pengirim mem-publish pesan ke sebuah channel (nama), tanpa tahu siapa yang mendengar, dan setiap yang sedang subscribe ke channel itu menerimanya saat itu juga. Redis bisa jadi perantaranya (message broker) antar proses atau server. Bayangin pengeras suara di pasar: petugas mengumumkan "cucian #12 siap diambil" di saluran pengumuman, dan semua yang ada di pasar mendengar. Yang sedang pulang gak dengar apa-apa, karena pengumuman gak direkam. Kalau perlu rekaman, pakai buku catatan yang bisa dibaca belakangan: itu Redis Streams.

🧪 Tebakan cepat: Laundry Bu Rini makin ramai, jadi server realtime-nya sekarang dua proses Node: A di port 3000 dan B di port 3001, di belakang load balancer. Tablet Mas Eko tersambung ke A, HP Pak Darto kebetulan ke B. Proses A menjalankan io.to('order:12').emit('order:status', ...). Sampaikah kabarnya ke HP Pak Darto? Jawabannya ada di bagian "Satu proses cuma kenal socket miliknya".

Satu proses cuma kenal socket miliknya

Room di Socket.IO (pelajaran Rooms) disimpan di memori proses, sebagai Map dari nama room ke kumpulan socket. Proses A gak punya daftar socket milik B, dan gak ada jalur antara keduanya.

Dicoba dengan socket.io 4.8.4, dua server di dua port, masing-masing punya satu client di room order:12: emit dari A cuma sampai ke client yang tersambung ke A. HP Pak Darto di B gak menerima apa pun, dan gak ada error. Itu jawaban tebakannya. Dengan satu proses semuanya jalan, jadi masalah ini baru muncul saat aplikasinya dibesarkan.

Obatnya: setiap proses mengirim kabar ke perantara, dan setiap proses mendengarkan perantara itu lalu meneruskan ke socket miliknya sendiri.

PUBLISH dan SUBSCRIBE di Redis

Dua terminal redis-cli (Redis 7.2.7). Terminal 1 berlangganan:

127.0.0.1:6379> SUBSCRIBE papan
1) "subscribe"
2) "papan"
3) (integer) 1

Terminal 2 mengirim:

127.0.0.1:6379> PUBLISH papan '{"id":12,"status":"Siap diambil"}'
(integer) 1

Di terminal 1 muncul:

1) "message"
2) "papan"
3) "{\"id\":12,\"status\":\"Siap diambil\"}"

import { createClient } from 'redis' // redis 6.3.0

const pub = createClient({ url: 'redis://localhost:6379' })
const sub = pub.duplicate()                   // koneksi kedua, khusus mendengar
await pub.connect()
await sub.connect()

// setiap proses (A dan B) mendengar, lalu meneruskan ke socket miliknya
await sub.subscribe('papan', (pesan) => {
  const d = JSON.parse(pesan)
  io.to(`order:${d.id}`).emit('order:status', d)
})

// proses mana pun yang menerima perubahan dari Mas Eko
await pub.publish('papan', JSON.stringify({ id: 12, status: 'Siap diambil' }))

Channel dan pola

Channel cuma nama, dibuat saat dipakai. PSUBSCRIBE berlangganan ke pola. Terminal 1 menjalankan PSUBSCRIBE order:*, terminal 2 menjalankan PUBLISH order:12 '{"status":"Siap diambil"}' (balasannya (integer) 1). Yang diterima subscriber pola punya satu baris tambahan, polanya:

1) "pmessage"
2) "order:*"
3) "order:12"
4) "{\"status\":\"Siap diambil\"}"

Di Node: await sub.pSubscribe('order:*', (pesan, channel) => ...). Setiap PUBLISH dicocokkan ke semua pola yang terdaftar, jadi pakai pola seperlunya.

Kalau perlu memutar ulang: Redis Streams

Redis Streams adalah daftar pesan yang disimpan di Redis, masing-masing dengan id <milidetik>-<urutan>. Pembaca bisa mulai dari id mana pun, jadi yang tadi mati bisa melanjutkan dari id terakhir yang ia baca.

127.0.0.1:6379> XADD papan:log * id 12 status "Siap diambil"
"1791686460000-0"
127.0.0.1:6379> XREAD COUNT 10 STREAMS papan:log 0
1) 1) "papan:log"
   2) 1) 1) "1791686460000-0"
         2) 1) "id"
            2) "12"
            3) "status"
            4) "Siap diambil"

Pub/Sub untuk "siapa pun yang sedang mendengar, sekarang". Streams untuk "jangan sampai ada yang terlewat".

Socket.IO di banyak proses: Redis adapter

Gak perlu menulis jembatan sendiri. Adapter Socket.IO adalah bagian yang menyimpan room dan menyiarkan emit; adapter Redis meneruskan setiap emit lewat Redis.

import { createClient } from 'redis'
import { createAdapter } from '@socket.io/redis-adapter' // 8.3.0

const pub = createClient({ url: 'redis://localhost:6379' })
const sub = pub.duplicate()
await Promise.all([pub.connect(), sub.connect()])
io.adapter(createAdapter(pub, sub))

io.to('order:12').emit('order:status', { id: 12, status: 'Siap diambil' }) // sampai ke A dan B

⚠️ Jebakan yang sering ketemu

TL;DR: Room Socket.IO hidup di memori satu proses, jadi emit dari A gak sampai ke socket di B. Redis Pub/Sub jadi perantara antar proses: PUBLISH membalas jumlah penerima, SUBSCRIBE/PSUBSCRIBE (pola) mendengar, dan semuanya fire-and-forget: yang mati melewatkan pesan. Redis Streams (XADD, XREAD dari id terakhir, consumer group dengan XREADGROUP dan XACK) menyimpan pesan untuk diputar ulang. @socket.io/redis-adapter 8.3.0 menyambungkan semua proses lewat Pub/Sub (tanpa recovery); @socket.io/redis-streams-adapter 0.3.1 lewat stream, dengan connection state recovery.

Yang akan kamu pelajari