Posted inDevelopment

Xử lý trùng lặp tin nhắn trong RabbitMQ: Triển khai Idempotent Consumer với Node.js

Development tutorial - IT technology blog
Development tutorial - IT technology blog

Bài toán kinh điển: Tại sao khách hàng bị trừ tiền hai lần?

Hãy tưởng tượng bạn đang vận hành một hệ thống thanh toán. Khách hàng bấm nút xác nhận, nhưng do mạng lag (network jitter), Consumer xử lý xong mà chưa kịp gửi ACK về cho RabbitMQ thì kết nối đứt. RabbitMQ thấy mất kết nối nên đẩy lại tin nhắn đó cho một Consumer khác. Kết quả? Khách hàng bị trừ tiền hai lần. Đây không phải lỗi code, đây là đặc tính của hệ thống phân tán.

RabbitMQ cam kết “at-least-once delivery” (chắc chắn nhận được ít nhất một lần), nhưng không đảm bảo “exactly-once”. Việc trùng lặp tin nhắn là điều tất yếu. Để giải quyết, chúng ta cần biến Consumer thành một Idempotent Consumer. Hiểu đơn giản: Dù nhận một tin nhắn 1 lần hay 100 lần, trạng thái hệ thống vẫn không thay đổi.

Chuẩn bị môi trường

Để thực hành, bạn cần Node.js và Docker để chạy nhanh RabbitMQ cùng Redis. Chúng ta dùng Redis làm kho lưu trữ message_id vì tốc độ đọc/ghi cực nhanh, phù hợp để check trùng lặp trong vài mili giây.

# Chạy nhanh RabbitMQ và Redis
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
docker run -d --name redis -p 6379:6379 redis:alpine

Khởi tạo project và cài đặt các thư viện cần thiết:

mkdir rabbitmq-idempotency && cd rabbitmq-idempotency
npm init -y
npm install amqplib ioredis uuid

Chiến thuật: Kiểm tra trước, xử lý sau

Quy trình rất rõ ràng: Mỗi tin nhắn gửi đi phải đính kèm một messageId duy nhất. Khi Consumer nhận được, nó sẽ tra cứu trong Redis. Nếu ID đã tồn tại, bỏ qua ngay. Nếu chưa, tiến hành xử lý và lưu ID đó vào Redis.

1. Producer: Gửi tin nhắn kèm định danh

Tuyệt đối không gửi payload “trần trụi”. Hãy bọc dữ liệu vào một object có metadata để dễ quản lý.

const amqp = require('amqplib');
const { v4: uuidv4 } = require('uuid');

async function sendOrder() {
    const conn = await amqp.connect('amqp://localhost');
    const channel = await conn.createChannel();
    const queue = 'order_queue';

    const message = {
        id: uuidv4(), // Định danh duy nhất cho mỗi transaction
        data: { orderId: 'ORD-999', amount: 500000 }
    };

    await channel.assertQueue(queue, { durable: true });
    channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)), {
        persistent: true
    });

    console.log(`[x] Đã gửi đơn hàng: ${message.id}`);
    setTimeout(() => conn.close(), 500);
}

sendOrder();

2. Consumer: Cơ chế chống trùng lặp với SETNX

Tại đây, chúng ta dùng lệnh SETNX (Set if Not Exists) của Redis. Đây là thao tác nguyên tử (atomic). Nó giúp đảm bảo dù 10 instance cùng nhận một message, chỉ duy nhất một instance chiếm được quyền xử lý.

const amqp = require('amqplib');
const Redis = require('ioredis');
const redis = new Redis();

async function consume() {
    const conn = await amqp.connect('amqp://localhost');
    const channel = await conn.createChannel();
    const queue = 'order_queue';

    await channel.assertQueue(queue, { durable: true });
    channel.prefetch(1);

    channel.consume(queue, async (msg) => {
        if (!msg) return;
        
        const { id, data } = JSON.parse(msg.content.toString());
        
        // Thử ghi vào Redis với TTL 24 giờ để tránh rác bộ nhớ
        const isNew = await redis.set(`msg:${id}`, 'processing', 'NX', 'EX', 86400);

        if (isNew) {
            try {
                console.log(`[v] Đang xử lý message: ${id}`);
                await processOrder(data); // Giả lập logic nghiệp vụ
                channel.ack(msg);
            } catch (err) {
                console.error("Lỗi xử lý:", err);
                await redis.del(`msg:${id}`); // Xóa key để có thể retry
                channel.nack(msg, false, true);
            }
        } else {
            console.warn(`[!] Phát hiện trùng lặp: ${id}. Bỏ qua...`);
            channel.ack(msg); // Vẫn phải ACK để xóa khỏi queue
        }
    });
}

async function processOrder(data) {
    return new Promise(res => setTimeout(res, 1000));
}

consume();

Trong thực tế, mình từng vận hành hệ thống xử lý hơn 1 triệu message/ngày. Bài học rút ra là TTL (Time To Live) cực kỳ quan trọng. Nếu không đặt TTL, Redis sẽ phình to khủng khiếp sau vài tháng, gây lãng phí tài nguyên và làm chậm tốc độ tra cứu.

Giám sát và đo lường

Đừng chỉ code xong rồi để đó. Bạn cần quan tâm đến các chỉ số sau:

  • Redis Hit/Miss Rate: Nếu tỉ lệ trùng lặp đột ngột tăng vọt (ví dụ > 5%), có thể hệ thống mạng đang gặp vấn đề nghiêm trọng.
  • Consumer Lag: Theo dõi trên RabbitMQ UI. Nếu Unacknowledged Messages tăng cao, có nghĩa là logic xử lý của bạn đang chậm hơn tốc độ tin nhắn đổ về.
  • Dead Letter Queue (DLQ): Luôn chuẩn bị một “nghĩa địa” cho các tin nhắn lỗi quá nhiều lần. Đừng để một tin nhắn lỗi làm nghẽn toàn bộ luồng xử lý.

Xây dựng hệ thống phân tán là học cách chấp nhận sự bất định. Thay vì cố gắng ngăn chặn trùng lặp tuyệt đối, hãy thiết kế Consumer sao cho nó đủ thông minh để nhận biết và từ chối những gì nó đã làm rồi.

Share:
Tags: