Message Queues — RabbitMQ, Kafka Cho Hệ Thống Phân Tán • blog.dopana

Message queue là xương sống của hệ thống phân tán. Nó cho phép các service giao tiếp bất đồng bộ, chịu lỗi và scale độc lập.

Khi Nào Cần Message Queue?#

Decoupling — Service A không cần biết Service B
Buffering — Chịu được traffic spike (queue làm đệm)
Retry — Nếu worker fail, message tự động retry
Fan-out — Một message đến nhiều consumer

RabbitMQ — Message Broker Cổ Điển#

Concepts#

Producer → Exchange → Queue → Consumer
              │
        (routing key, binding)

txt

Exchange — nhận message từ producer, route đến queue
Queue — lưu message, consumer đọc từ queue
Binding — kết nối exchange → queue

Node.js với amqplib#

import amqp from 'amqplib';

// Producer
async function sendMessage(queue: string, message: object) {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  await channel.assertQueue(queue, { durable: true });
  channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)), {
    persistent: true, // Lưu xuống disk
  });
  console.log(`Sent: ${JSON.stringify(message)}`);
}

// Consumer
async function consumeMessages(queue: string) {
  const conn = await amqp.connect('amqp://localhost');
  const channel = await conn.createChannel();
  await channel.assertQueue(queue, { durable: true });

  channel.consume(queue, (msg) => {
    if (msg) {
      const data = JSON.parse(msg.content.toString());
      console.log(`Received: ${data}`);

      // Acknowledge — xác nhận đã xử lý xong
      channel.ack(msg);
    }
  });
}

// Usage
await sendMessage('order.created', { orderId: 123, userId: 456 });
await consumeMessages('order.created');

typescript

Exchange Types#

// Direct — route theo routing key chính xác
await channel.assertExchange('direct_logs', 'direct');
await channel.bindQueue('queue_errors', 'direct_logs', 'error');

// Topic — route theo pattern
await channel.assertExchange('topic_logs', 'topic');
await channel.bindQueue('queue_all', 'topic_logs', '*.critical');
await channel.bindQueue('queue_errors', 'topic_logs', 'error.#');

// Fanout — broadcast đến tất cả queue
await channel.assertExchange('fanout_logs', 'fanout');
// Tất cả queue bind đều nhận được message

typescript

Kafka — Event Streaming Platform#

Concepts#

Producer → Topic (Partition 0, 1, 2…) → Consumer Group

Topic — category của message
Partition — chia topic để song song
Consumer Group — nhiều consumer cùng group chia nhau partition
Offset — vị trí message trong partition

Node.js với kafkajs#

import { Kafka } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092'],
});

// Producer
async function produce() {
  const producer = kafka.producer();
  await producer.connect();

  await producer.send({
    topic: 'order-events',
    messages: [
      { key: 'order_123', value: JSON.stringify({ id: 123, status: 'created' }) },
      { key: 'order_456', value: JSON.stringify({ id: 456, status: 'paid' }) },
    ],
  });

  await producer.disconnect();
}

// Consumer
async function consume() {
  const consumer = kafka.consumer({ groupId: 'order-service' });
  await consumer.connect();
  await consumer.subscribe({ topic: 'order-events', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        partition,
        offset: message.offset,
        value: message.value?.toString(),
      });

      // auto-commit offset (hoặc manual)
    },
  });
}

typescript

Kafka vs RabbitMQ#

	RabbitMQ	Kafka
Model	Smart broker, dumb consumer	Dumb broker, smart consumer
Storage	Xoá sau khi consumed	Giữ message lâu (configurable)
Performance	~10K msg/s	~1M msg/s
Message ordering	Per queue	Per partition
Use case	Task distribution	Event streaming, log
Replay	❌	✅ (từ offset cũ)

NATS — Lightweight, Đơn Giản#

import { connect } from 'nats';

const nc = await connect({ servers: 'nats://localhost:4222' });

// Publish
nc.publish('order.created', JSON.stringify({ id: 123 }));

// Subscribe
const sub = nc.subscribe('order.created');
for await (const msg of sub) {
  console.log(msg.data.toString());
}

// Request-Reply
const response = await nc.request('user.get', JSON.stringify({ id: 1 }));

typescript

NATS cực kỳ nhẹ — latency micro giây.

Pattern Thực Tế#

1. Event Sourcing#

Mọi thay đổi state đều là event:

// Order service
await producer.send({
  topic: 'order-events',
  messages: [
    { value: JSON.stringify({ type: 'OrderCreated', data: order }) },
    { value: JSON.stringify({ type: 'OrderPaid', data: payment }) },
    { value: JSON.stringify({ type: 'OrderShipped', data: shipment }) },
  ],
});

// Notification service consume events
// → Gửi email khi OrderCreated
// → Gửi SMS khi OrderShipped

typescript

2. Competing Consumers#

Nhiều worker cùng xử lý queue:

Queue: [msg1, msg2, msg3, msg4, msg5]
         ↓      ↓      ↓      ↓      ↓
      Worker1 Worker2 Worker3 Worker1 Worker2

txt

Tự động scale — thêm worker = xử lý nhanh hơn.

3. Dead Letter Queue (DLQ)#

Message không xử lý được → chuyển sang DLQ:

channel.consume(queue, (msg) => {
  try {
    processMessage(msg);
    channel.ack(msg);
  } catch (err) {
    const retries = msg.properties.headers['x-retry'] || 0;
    if (retries < 3) {
      channel.nack(msg, false, true); // Requeue
    } else {
      channel.nack(msg, false, false); // → DLQ
    }
  }
});

typescript

Docker Compose#

services:
  rabbitmq:
    image: rabbitmq:4-management
    ports:
      - "5672:5672"  # AMQP
      - "15672:15672" # Management UI

  kafka:
    image: confluentinc/cp-kafka:latest
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

  nats:
    image: nats:latest
    ports:
      - "4222:4222"

yaml

Kết Luận#

RabbitMQ — task distribution, job queue, cần routing phức tạp
Kafka — event streaming, log, cần replay message
NATS — real-time, lightweight, micro giây latency

Chọn theo nhu cầu:

Đơn giản, cần retry → RabbitMQ
High throughput, event sourcing → Kafka
Siêu nhanh, real-time → NATS

Message Queues — RabbitMQ, Kafka Cho Hệ Thống Phân Tán

Khi Nào Cần Message Queue?#

RabbitMQ — Message Broker Cổ Điển#

Concepts#

Node.js với amqplib#

Exchange Types#

Kafka — Event Streaming Platform#

Concepts#

Node.js với kafkajs#

Kafka vs RabbitMQ#

NATS — Lightweight, Đơn Giản#

Pattern Thực Tế#

1. Event Sourcing#

2. Competing Consumers#

3. Dead Letter Queue (DLQ)#

Docker Compose#

Kết Luận#

Tài liệu tham khảo#