Arsitektur Event-Driven Skala Besar Menggunakan Redis Streams dan Node.js

1. Konteks Industri & Analisis Akar Masalah

Dalam lanskap arsitektur sistem terdistribusi modern, pergeseran dari komunikasi synchronous request-response (seperti REST API tradisional berbasis HTTP) menuju Asynchronous Event-Driven Architecture (EDA) bukan lagi sebuah kemewahan, melainkan keharusan mutlak. Saat aplikasi web Anda menghadapi lonjakan traffic hingga puluhan ribu request per detik, arsitektur monolitik konvensional atau pemanggilan layanan mikro secara langsung (direct RPC) akan mengalami bottleneck parah.

Akar masalah dari kegagalan sistem berskala besar sering kali berakar pada kopling ketat (tight coupling). Ketika sebuah layanan hulu (upstream service) mengirimkan data secara langsung ke layanan hilir (downstream service), kegagalan sesaat pada layanan hilir akan memicu efek domino (cascade failure) yang melumpuhkan seluruh sistem. Selain itu, operasi I/O yang memblokir thread utama pada backend seperti Node.js akan menghabiskan alokasi memori dan CPU, memicu lonjakan latensi (latency spikes), dan berujung pada Out of Memory (OOM) killer di lingkungan sistem operasi.

Sebelum penerapan Event-Driven Architecture berbasis log-based message broker seperti Redis Streams, perusahaan kerap mengandalkan antrean sederhana berbasis list (LPUSH/RPOP) atau mekanisme Pub/Sub native Redis. Pub/Sub bawaan Redis bersifat fire-and-forget; jika consumer terputus sesaat dari jaringan, pesan melayang begitu saja dan hilang selamanya tanpa jejak. Antrean berbasis list tradisional juga rentan terhadap masalah lost update ketika beberapa worker mencoba mengambil pesan yang sama tanpa manajemen state consumer group yang terstruktur.

Setelah mengadopsi Redis Streams yang dipadukan dengan performa asynchronous non-blocking I/O dari Node.js, sistem mendapatkan kemampuan penyerapan beban (load levelling) yang luar biasa. Pesan tidak lagi hilang jika consumer down, karena tersimpan rapi dalam log terstruktur di memori (dengan opsi persistensi disk AOF/RDB). Node.js worker dapat memproses event secara paralel menggunakan Consumer Groups, menjamin at-least-once delivery, dan menyediakan mekanisme pelacakan posisi baca (ack/nack) yang presisi.

[Client Requests] ---> [Nginx Reverse Proxy] ---> [Node.js API Gateway]
                                                           |
                                                (XADD Event to Stream)
                                                           v
                                                   [Redis Streams Engine]
                                                      /              \
                   (Consumer Group Workers)          /                \
                   [Node.js Worker Node A]          /                  \  [Node.js Worker Node B]
                          |                        /                    |
                          v                       v                     v
                   [MySQL / Primary DB]    [S3 Object Store]    [Postfix SMTP Relay]

2. Konsep Inti & Deep Dive Arsitektur

Redis Streams, yang diperkenalkan sejak Redis 5.0, menghadirkan struktur data menyerupai append-only log namun dengan kapabilitas manajemen pengiriman pesan tingkat lanjut, terinspirasi secara konseptual dari Apache Kafka. Berbeda dengan struktur data Redis lainnya seperti List atau Sorted Set, Redis Streams menyimpan setiap entri sebagai pasangan field-value yang unik dengan ID berbasis waktu (contoh: 1689000000000-0).

Anatomi Redis Stream

Setiap pesan di dalam stream memiliki:

1. Stream Key: Nama log tempat pesan disimpan (misal: order:events).
2. Entry ID: Kombinasi millisecond timestamp dan sequence number (misal: 1689123456789-0). Timestamp menjamin urutan waktu alami, sementara sequence number memastikan keunikan ID meskipun beberapa pesan masuk dalam milidetik yang sama.
3. Field-Value Pairs: Data aktual yang dibawa oleh event dalam format dictionary.

Consumer Groups

Fitur paling krusial untuk arsitektur skala besar adalah Consumer Groups. Sebuah Consumer Group adalah abstraksi di atas sebuah stream tunggal yang memungkinkan sekumpulan consumer (dalam hal ini instance aplikasi Node.js Anda) untuk membagi beban kerja pemrosesan pesan.

• Pending Entries List (PEL): Redis mencatat setiap pesan yang telah dikirimkan ke sebuah consumer di dalam group tetapi belum diakui (acknowledged). Hal ini memastikan bahwa jika sebuah worker Node.js mengalami crash atau segmentation fault di tengah jalan, pesan tersebut tidak hilang dan dapat dipulihkan (claimed) oleh worker lain melalui perintah XCLAIM atau XAUTOCLAIM.
• Acknowledgment (XACK): Mekanisme eksplisit yang memberi tahu Redis bahwa sebuah pesan telah selesai diproses dengan sukses oleh consumer, sehingga pesan tersebut dapat dihapus dari PEL.

Stream: order:events
[ID: 1001] -> [ID: 1002] -> [ID: 1003] -> [ID: 1004] -> [ID: 1005]
     ^            ^            ^            ^            ^
     |            |            |            |            |
+-----------------------------------------------------------------+
|                       CONSUMER GROUP: "billing-service"         |
|                                                                 |
|  +-----------------------+              +--------------------+
|  | Consumer Node #1      |              | Consumer Node #2   |
|  | (Processing ID: 1001) |              | (Processing ID: 1002)|
|  +-----------------------+              +--------------------+
+-----------------------------------------------------------------+

3. Perbandingan Komparatif / Analisis Pendekatan

Untuk menentukan kapan harus menggunakan Redis Streams ketimbang solusi messaging broker lain dalam ekosistem infrastruktur Anda, perhatikan matriks perbandingan komprehensif di bawah ini:

[!IMPORTANT] Jangan pernah menggunakan Redis Pub/Sub untuk arsitektur task queue atau pemrosesan transaksi finansial. Gunakan Redis Streams atau RabbitMQ/Kafka jika Anda membutuhkan keandalan delivery dan proteksi terhadap kegagalan worker.

4. Prasyarat Sistem & Persiapan Lingkungan

Sebelum mulai menulis kode implementasi, pastikan infrastruktur sistem operasi dan runtime environment Anda telah memenuhi spesifikasi produksi berikut:

• OS: Ubuntu Server 22.04 LTS / Debian 12 Bookworm
• Node.js: Versi 20.x LTS (atau yang lebih baru dengan dukungan native ES Modules dan worker threads)
• Redis Server: Versi 7.0+ (direkomendasikan untuk fitur XAUTOCLAIM yang optimal)
• Web Server / Process Manager: PM2 atau systemd untuk menjaga proses worker Node.js tetap hidup (daemonization).

Instalasi & Konfigurasi Redis di Linux

Perbarui sistem dan instal Redis server dari repositori resmi:

sudo apt update && sudo apt upgrade -y
sudo apt install redis-server -y

Konfigurasikan file /etc/redis/redis.conf untuk mengaktifkan persistensi tingkat lanjut dan mengamankan instance dengan password:

# /etc/redis/redis.conf
bind 127.0.0.1
port 6379
protected-mode yes
requirepass "S3cr3tR3d1sP@ssw0rd2024!"

# Aktifkan append-only file untuk durabilitas data maksimal
appendonly yes
appendfsync everysec

# Batasi penggunaan memori maksimal
maxmemory 2gb
maxmemory-policy noeviction

Restart layanan Redis untuk menerapkan konfigurasi:

sudo systemctl restart redis-server
sudo systemctl enable redis-server

5. Panduan Implementasi Langkah demi Langkah

Kita akan membangun arsitektur dua bagian: Producer (yang menerima request web via Express.js dan menerbitkan event ke Redis Stream) dan Consumer (worker Node.js background yang membaca event, memprosesnya, dan menangani error secara aman).

Struktur Direktori Project

redis-event-system/
├── package.json
├── .env
├── producer.js
└── consumer.js

Inisialisasi Project & Dependencies

Buat direktori baru dan instal library ioredis (pilihan klien Redis terbaik di Node.js untuk dukungan penuh Redis Streams):

mkdir redis-event-system && cd redis-event-system
npm init -y
npm install ioredis dotenv express

1. File Environment (.env)

PORT=3000
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_PASSWORD=S3cr3tR3d1sP@ssw0rd2024!
STREAM_NAME=ecommerce:orders
CONSUMER_GROUP=order-processing-group

2. Implementasi Producer (producer.js)

Producer bertindak sebagai titik masuk web service yang menerima transaksi, memvalidasi payload, dan memasukkannya ke dalam Redis Stream.

// producer.js
import express from 'express';
import Redis from 'ioredis';
import dotenv from 'dotenv';

dotenv.config();

const app = express();
app.use(express.json());

const redis = new Redis({
  host: process.env.REDIS_HOST,
  port: process.env.REDIS_PORT,
  password: process.env.REDIS_PASSWORD,
  maxRetriesPerRequest: null,
});

const STREAM_NAME = process.env.STREAM_NAME;

app.post('/api/orders', async (req, res) => {
  try {
    const { productId, quantity, customerId } = req.body;

    if (!productId || !quantity || !customerId) {
      return res.status(400.json({ error: 'Payload tidak lengkap' }));
    }

    const payload = {
      productId,
      quantity: String(quantity),
      customerId,
      timestamp: String(Date.now()),
    };

    // Menambahkan pesan ke Redis Stream
    // XADD stream-name * field1 value1 field2 value2 ...
    const messageId = await redis.xadd(
      STREAM_NAME,
      '*',
      'productId',
      payload.productId,
      'quantity',
      payload.quantity,
      'customerId',
      payload.customerId,
      'timestamp',
      payload.timestamp
    );

    console.log(`[Producer] Event berhasil diterbitkan dengan ID: ${messageId}`);

    return res.status(202).json({
      status: 'Accepted',
      message: 'Order sedang diproses dalam background queue',
      eventId: messageId,
    });
  } catch (error) {
    console.error('[Producer Error] Gagal menerbitkan event:', error);
    return res.status(500).json({ error: 'Internal Server Error' });
  }
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`[Producer] Berjalan di port ${PORT}`);
});

3. Implementasi Consumer & Worker dengan DLQ (consumer.js)

Consumer menjalankan polling asinkron menggunakan perintah XREADGROUP, memproses event, melakukan acknowledgment, serta memindahkan pesan yang gagal berulang kali ke Dead Letter Queue (DLQ).

// consumer.js
import Redis from 'ioredis';
import dotenv from 'dotenv';

dotenv.config();

const redis = new Redis({
  host: process.env.REDIS_HOST,
  port: process.env.REDIS_PORT,
  password: process.env.REDIS_PASSWORD,
  maxRetriesPerRequest: null,
});

const STREAM_NAME = process.env.STREAM_NAME;
const CONSUMER_GROUP = process.env.CONSUMER_GROUP;
const CONSUMER_NAME = `worker-${process.pid}`;
const DLQ_STREAM = `${STREAM_NAME}:dlq`;
const MAX_RETRIES = 3;

async function setupConsumerGroup() {
  try {
    // Membuat consumer group, '0' berarti mulai dari awal stream jika group belum ada
    await redis.xgroup('CREATE', STREAM_NAME, CONSUMER_GROUP, '0', 'MKSTREAM');
    console.log(`[Consumer] Consumer Group '${CONSUMER_GROUP}' berhasil dibuat.`);
  } catch (error) {
    if (error.message.includes('BUSYGROUP')) {
      console.log(`[Consumer] Consumer Group '${CONSUMER_GROUP}' sudah ada, melanjutkan...`);
    } else {
      console.error('[Consumer Setup Error]:', error);
      process.exit(1);
    }
  }
}

async function processOrder(fields) {
  const data = {};
  for (let i = 0; i < fields.length; i += 2) {
    data[fields[i]] = fields[i + 1];
  }

  console.log(`[Worker ${CONSUMER_NAME}] Memproses pesanan untuk Produk: ${data.productId}, Customer: ${data.customerId}`);

  // Simulasi logika bisnis (misal: potong stok database / panggil payment gateway)
  if (data.productId === 'FAIL_TEST') {
    throw new Error('Simulasi kegagalan sistem pembayaran');
  }

  // Simulasi latensi proses
  await new Promise((resolve) => setTimeout(resolve, 1000));
  console.log(`[Worker ${CONSUMER_NAME}] Pesanan berhasil diselesaikan.`);
}

async function moveToDLQ(messageId, fields, errorMessage) {
  console.warn(`[DLQ] Pesanan ${messageId} gagal setelah max retries. Mengirim ke DLQ...`);
  
  // Masukkan ke stream Dead Letter Queue
  const dlqPayload = [...fields, 'errorReason', errorMessage, 'originalId', messageId];
  await redis.xadd(DLQ_STREAM, '*', ...dlqPayload);
  
  // Acknowledge pesan dari stream utama agar tidak diproses lagi
  await redis.xack(STREAM_NAME, CONSUMER_GROUP, messageId);
}

async function listenForEvents() {
  await setupConsumerGroup();

  console.log(`[Consumer] Worker ${CONSUMER_NAME} mendengarkan stream: ${STREAM_NAME}...`);

  while (true) {
    try {
      // XREADGROUP GROUP group consumer COUNT count BLOCK timeout STREAMS key id
      // '>' berarti ambil pesan baru yang belum pernah dikirim ke consumer manapun dalam group
      const response = await redis.xreadgroup(
        'GROUP',
        CONSUMER_GROUP,
        CONSUMER_NAME,
        'COUNT',
        1,
        'BLOCK',
        5000,
        'STREAMS',
        STREAM_NAME,
        '>'
      );

      if (!response) continue;

      const [stream, entries] = response[0];

      for (const [messageId, fields] of entries) {
        let attempts = 0;
        let processed = false;

        while (attempts < MAX_RETRIES && !processed) {
          try {
            attempts++;
            await processOrder(fields);
            
            // Jika sukses, kirim XACK
            await redis.xack(STREAM_NAME, CONSUMER_GROUP, messageId);
            processed = true;
          } catch (err) {
            console.error(`[Attempt ${attempts}/${MAX_RETRIES}] Gagal memproses pesan ${messageId}: ${err.message}`);
            if (attempts >= MAX_RETRIES) {
              await moveToDLQ(messageId, fields, err.message);
            } else {
              // Jeda sejenak sebelum mencoba ulang (exponential backoff sederhana)
              await new Promise((res) => setTimeout(res, 1000 * attempts));
            }
          }
        }
      }
    } catch (err) {
      console.error('[Stream Polling Error]:', err);
      await new Promise((res) => setTimeout(res, 2000));
    }
  }
}

listenForEvents().catch((err) => {
  console.error('[Fatal Error di Consumer Loop]:', err);
  process.exit(1);
});

6. Bedah Kode & Penjelasan Detail Baris Kunci

Mari kita bedah beberapa baris kode krusial yang menentukan ketahanan (resilience) dari arsitektur di atas:

1. maxRetriesPerRequest: null pada Klien Redis (ioredis): Secara default, ioredis akan mencoba menghubungkan ulang dan melempar error jika antrean perintah menumpuk saat koneksi terputus. Mengatur nilai ini ke null sangat penting untuk operasi blocking stream (XREADGROUP dengan parameter BLOCK) agar tidak terjadi Unhandled Promise Rejection ketika Redis mengalami restart singkat.

2. Parameter '>' pada Perintah XREADGROUP: Simbol khusus > menginstruksikan Redis untuk hanya memberikan pesan-pesan baru yang belum pernah dialokasikan ke consumer manapun di dalam consumer group yang sama. Ini menjamin bahwa jika Anda menjalankan 5 instance worker Node.js secara paralel, setiap pesan hanya akan diproses oleh tepat satu worker, mencegah duplikasi transaksi.

3. Mekanisme Dead Letter Queue (DLQ): Ketika suatu pesan gagal diproses berulang kali (melebihi MAX_RETRIES = 3), pesan tersebut tidak dibiarkan menggantung di dalam Pending Entries List (PEL) selamanya. Kode secara eksplisit menyalin data tersebut ke stream terpisah (ecommerce:orders:dlq) disertai alasan error, lalu melakukan XACK pada stream utama. Hal ini krusial agar Pending List tidak membengkak yang dapat menyebabkan memori Redis habis (Memory Leak di level sistem).

7. Skenario Produksi & Studi Kasus Dunia Nyata

Di lingkungan produksi berskala besar (misal: Flash Sale e-commerce dengan 50.000 pesanan per menit), arsitektur ini mendistribusikan beban secara efisien:

• Pemisahan Peran (Segregation of Concerns): Nginx bertindak sebagai reverse proxy dan SSL termination. Request masuk diterima oleh kluster Node.js API Gateway yang sangat ringan, langsung melakukan validate and dump ke Redis Streams dalam waktu di bawah 2 milidetik (XADD), kemudian langsung mengembalikan respons HTTP 202 Accepted kepada klien. Klien tidak perlu menunggu proses database yang berat.
• Horizontal Scaling Worker: Di latar belakang, Anda dapat menjalankan puluhan worker nodes menggunakan PM2 atau Kubernetes Pods yang mendengarkan Consumer Group yang sama (order-processing-group). Jika traffic meningkat mendadak, Anda cukup menaikkan jumlah replika kontainer worker tanpa mengubah satu baris pun kode pada API Gateway.

# Menjalankan 4 instance worker secara paralel menggunakan PM2
pm2 start consumer.js -i 4 --name "order-worker"

8. Common Pitfalls, Anti-Patterns & Solusi

Banyak engineer terjebak dalam jebakan klasik saat pertama kali mengimplementasikan sistem event-driven berbasis Redis:

• Anti-Pattern 1: Mengabaikan Pengelolaan Pending Entries List (PEL)

  • Masalah: Consumer crash, pesan masuk ke PEL, namun tidak pernah di-claim ulang oleh worker lain. Seiring waktu, PEL menumpuk dan menghabiskan RAM Redis.
  • Solusi: Implementasikan background sweep job rutin menggunakan perintah XAUTOCLAIM untuk mengambil alih pesan yang terbengkalai (idle) lebih dari durasi tertentu (misal: 10 menit).

• Anti-Pattern 2: Menggunakan Stream Key Tanpa Batasan Ukuran (Capping)

  • Masalah: Stream tumbuh secara eksponensial tanpa batas, menyebabkan memori server Redis habis (OOM).
  • Solusi: Gunakan argumen MAXLEN pada perintah XADD untuk memangkas stream secara otomatis secara approximate (contoh: MAXLEN ~ 100000).

9. Benchmark, Metrik Kinerja & Tuning Performa

Untuk menguji ketahanan sistem yang telah dibangun, kita dapat menggunakan tool benchmark HTTP populer seperti k6 atau autocannon untuk memicu load testing pada endpoint Producer.

# Instal autocannon secara global
npm install -g autocannon

# Lakukan uji beban dengan 100 koneksi simultan selama 30 detik
autocannon -c 100 -d 30 -m POST -H "Content-Type: application/json" -b '{"productId":"PROD-99","quantity":2,"customerId":"CUST-123\