Vận Hành Redis Streams Trong Production: Consumer Groups, Khôi Phục PEL Và Dead-Letter Queue

Cạm Bẫy Về Độ Tin Cậy Của Redis Pub/Sub

Rất nhiều đội ngũ kỹ thuật lựa chọn Redis Pub/Sub khi cần truyền phát thông điệp thời gian thực giữa các microservices. Mặc dù Pub/Sub mang lại tốc độ cực nhanh và mức tiêu thụ bộ nhớ tối thiểu, mô hình này hoạt động hoàn toàn theo nguyên lý 'bắn rồi quên' (fire-and-forget). Nếu subscriber bị ngắt kết nối do quá trình deploy, dừng để thu gom rác (GC pause) hoặc phân mảnh mạng (network partition), toàn bộ thông điệp được xuất bản trong khoảng thời gian đó sẽ biến mất vĩnh viễn. Không có lưu trữ bền vững, không có cơ chế xác nhận (acknowledgment), và không thể đọc lại dữ liệu lịch sử.

Redis Streams (ra mắt từ Redis 5.0) giải quyết triệt để điểm yếu này bằng cấu trúc log chỉ cho phép ghi tiếp (append-only) được lưu trữ bền vững trên đĩa tương tự như Apache Kafka, nhưng vẫn giữ được tốc độ truy xuất trên RAM dưới một mili-giây. Tuy nhiên, chỉ đơn thuần chuyển đổi từ PUBLISH sang XADD là chưa đủ. Để xây dựng hệ thống xử lý sự kiện chuẩn production, bạn cần làm chủ mô hình consumer group, quản trị danh sách thông điệp chờ (Pending Entries List - PEL) và thiết kế cơ chế dead-letter queue hiệu quả.

Cấu Trúc Cốt Lõi: Entries, Consumer Groups Và PEL

Khi đẩy sự kiện vào stream bằng XADD, Redis tự động gán một mã định danh tăng dần gồm timestamp mili-giây và số thứ tự (ví dụ: 1712839201942-0). Khi đọc các bản ghi này thông qua consumer group bằng lệnh XREADGROUP, Redis theo dõi trạng thái phân phối thông qua hai cấu trúc dữ liệu nội bộ:

• Last Delivered ID: Con trỏ đánh dấu mốc sự kiện mới nhất đã được phân phối cho bất kỳ consumer nào trong group.
• Pending Entries List (PEL): Bảng theo dõi riêng cho từng consumer, ghi nhận mọi thông điệp đã được lấy về nhưng chưa được xác nhận hoàn tất qua lệnh XACK. Mỗi bản ghi lưu trữ ID tin nhắn, tên consumer, số lần phân phối (delivery count) và thời gian nhàn rỗi (idle time).

Nếu một worker gặp sự cố (crash) khi đang xử lý, thông điệp đó vẫn nằm kẹt trong PEL của worker đó. Nếu không có worker khác chủ động thu hồi (claim), sự kiện đó sẽ bị đình trệ vô thời hạn.

Bước 1: Ghi Nhận Sự Kiện Và Kiểm Soát Dung Lượng Stream

Một stream tăng trưởng không giới hạn sẽ sớm làm cạn kiệt RAM của máy chủ. Khi ghi sự kiện, hãy sử dụng toán tử cắt tỉa xấp xỉ (~) kết hợp với MAXLEN. Cách này giúp Redis giải phóng bộ nhớ theo từng node của cây radix thay vì phải tính toán cắt bỏ từng bản ghi riêng lẻ:

// Ghi sự kiện đơn hàng và giữ kích thước stream xấp xỉ 100.000 bản ghi
XADD orders_stream MAXLEN ~ 100000 * order_id 98421 user_id 4012 amount 149.99 currency USD status CREATED

Tiếp theo, khởi tạo consumer group. Thiết lập ID bắt đầu là $ nếu muốn group chỉ tiếp nhận các sự kiện mới phát sinh sau thời điểm tạo, hoặc dùng 0 nếu muốn đọc lại toàn bộ lịch sử stream:

// Tạo group 'order_processors' đọc từ mốc mới nhất của 'orders_stream'
// MKSTREAM tự động tạo key stream nếu chưa tồn tại
XGROUP CREATE orders_stream order_processors $ MKSTREAM

Bước 2: Xây Dựng Consumer Loop Chịu Lỗi

Một worker chạy production cần xử lý hai luồng công việc song song: tiếp nhận thông điệp mới và thu hồi các thông điệp bị bỏ rơi từ những worker bị crash. Mã nguồn Node.js dưới đây sử dụng thư viện ioredis triển khai đầy đủ chu trình này:

import Redis from "ioredis";

const redis = new Redis({ host: "127.0.0.1", port: 6379 });
const STREAM = "orders_stream";
const GROUP = "order_processors";
const CONSUMER_NAME = `worker-${process.pid}`;
const MAX_RETRIES = 3;
const DLQ_STREAM = "orders_stream_dlq";

async function processOrder(messageId, payload) {
  // Giả lập logic xử lý nghiệp vụ
  if (Math.random() < 0.05) {
    throw new Error("Lỗi gián đoạn kết nối cơ sở dữ liệu giả lập");
  }
  console.log(`[${CONSUMER_NAME}] Đã xử lý ${messageId}:`, payload);
}

async function handleDeadLetter(messageId, payload, error) {
  await redis.xadd(
    DLQ_STREAM,
    "*",
    "original_id", messageId,
    "error", error.message,
    "failed_at", Date.now().toString(),
    "payload", JSON.stringify(payload)
  );
  await redis.xack(STREAM, GROUP, messageId);
  console.error(`[DLQ] Đã cô lập bản ghi lỗi ${messageId}`);
}

async function reclaimStaleMessages() {
  const MIN_IDLE_TIME_MS = 30000; // Nhàn rỗi quá 30 giây
  let cursor = "0-0";

  while (true) {
    // XAUTOCLAIM chuyển quyền sở hữu các entry bị treo một cách an toàn
    const [nextCursor, entries] = await redis.xautoclaim(
      STREAM,
      GROUP,
      CONSUMER_NAME,
      MIN_IDLE_TIME_MS,
      cursor,
      "COUNT", 50
    );

    for (const [id, fields] of entries) {
      const payload = Object.fromEntries(
        fields.reduce((acc, cur, i, arr) => {
          if (i % 2 === 0) acc.push([cur, arr[i + 1]]);
          return acc;
        }, [])
      );

      // Kiểm tra số lần thử lại thông qua XPENDING
      const [[, , , deliveryCount]] = await redis.xpending(
        STREAM, GROUP, id, id, 1
      );

      if (deliveryCount > MAX_RETRIES) {
        await handleDeadLetter(id, payload, new Error("Vượt quá giới hạn số lần thử lại"));
        continue;
      }

      try {
        await processOrder(id, payload);
        await redis.xack(STREAM, GROUP, id);
      } catch (err) {
        console.warn(`[RECLAIM-FAIL] Lần thử thứ ${deliveryCount} thất bại cho ${id}`);
      }
    }

    cursor = nextCursor;
    if (cursor === "0-0") break;
  }
}

async function startConsumer() {
  console.log(`Khởi động consumer ${CONSUMER_NAME}...`);

  // Định kỳ quét và thu hồi các thông điệp bị bỏ rơi
  setInterval(reclaimStaleMessages, 10000);

  while (true) {
    try {
      // Đọc các thông điệp hoàn toàn mới dành cho group
      const response = await redis.xreadgroup(
        "GROUP", GROUP, CONSUMER_NAME,
        "BLOCK", 2000,
        "COUNT", 10,
        "STREAMS", STREAM,
        ">"
      );

      if (!response) continue;

      const [, messages] = response[0];
      for (const [id, fields] of messages) {
        const payload = Object.fromEntries(
          fields.reduce((acc, cur, i, arr) => {
            if (i % 2 === 0) acc.push([cur, arr[i + 1]]);
            return acc;
          }, [])
        );

        try {
          await processOrder(id, payload);
          await redis.xack(STREAM, GROUP, id);
        } catch (err) {
          console.error(`Lỗi xử lý tại ${id}: ${err.message}`);
          // Không gọi XACK; bản ghi tiếp tục nằm trong PEL để thử lại hoặc XAUTOCLAIM
        }
      }
    } catch (loopError) {
      console.error("Lỗi vòng lặp consumer:", loopError);
      await new Promise((r) => setTimeout(r, 1000));
    }
  }
}

startConsumer();

Phân Tích Cơ Chế Kiến Trúc Trọng Tâm

1. Ý Nghĩa Của Các Ký Hiệu ID Đặc Biệt

Khi thao tác với consumer groups trong Redis Streams, hai ký hiệu ID mang ý nghĩa nghiệp vụ hoàn toàn khác nhau:

• >: Yêu cầu lấy các thông điệp chưa từng được phân phối cho bất kỳ worker nào trong group này.
• 0 hoặc một ID cụ thể: Đọc lại các thông điệp đang nằm trong trạng thái pending (chưa được XACK) thuộc quyền sở hữu của chính consumer đang gọi lệnh.

2. So Sánh XAUTOCLAIM Và XCLAIM

Trước phiên bản Redis 6.2, để thu hồi thông điệp mồ côi từ một worker bị sập, hệ thống phải thực hiện hai bước: gọi lệnh XPENDING để tìm danh sách ID nhàn rỗi, sau đó phát tiếp lệnh XCLAIM. Redis 6.2 ra mắt lệnh XAUTOCLAIM gộp toàn bộ quá trình quét, lọc theo thời gian nhàn rỗi và chuyển giao quyền sở hữu thành một thao tác nguyên tử O(1) cho mỗi phần tử, giảm thiểu đáng kể chi phí round-trip trên hạ tầng mạng.

3. Dead-Letter Queue (Hàng Đợi Xử Lý Lỗi)

Nếu không có trần giới hạn số lần thử lại, một bản ghi mang dữ liệu hỏng (poison pill) sẽ khiến các worker liên tục sập và thử lại vòng lặp đó mãi mãi. Bằng cách kiểm tra số lần phân phối (delivery count) từ PEL thông qua XPENDING, worker có thể phát hiện khi một bản ghi vượt quá ngưỡng quy định (ví dụ: quá 3 lần), chuyển tiếp thông điệp sang một stream DLQ riêng biệt và phát lệnh XACK để gỡ bỏ nó khỏi dòng xử lý chính.

Rủi Ro Vận Hành Và Tối Ưu Hóa Production

Cảnh báo: Việc quên gọi XACK không làm gián đoạn việc đọc các thông điệp mới ngay lập tức, nhưng nó sẽ âm thầm làm phình to bộ nhớ RAM của Redis theo thời gian. Mỗi bản ghi chưa xác nhận sẽ liên tục tiêu tốn metadata trong cây radix PEL nội bộ của group.

Những nguyên tắc vận hành cốt lõi cần tuân thủ khi đưa vào hệ thống thực tế:

1. Giám Sát Dung Lượng PEL: Theo dõi chỉ số này qua lệnh XINFO GROUPS orders_stream. Nếu con số pending liên tục tăng cao, điều đó phản ánh worker đang bị crash, xử lý quá chậm hoặc có luồng code bị thiếu XACK.
2. Xử Lý Tắt Tiến Trình Mềm Dẻo (Graceful Shutdown): Đăng ký lắng nghe tín hiệu SIGTERM. Hãy đợi các task đang xử lý dở hoàn tất và gửi lệnh XACK trước khi tiến trình tắt hẳn, tránh tạo ra tải dư thừa cho cơ chế XAUTOCLAIM.
3. Bắt Buộc Đảm Bảo Tính Bất Biến (Idempotency): Vì Redis Streams trong mô hình consumer group hoạt động theo cơ chế đảm bảo gửi ít nhất một lần (at-least-once), hiện tượng phân phối lặp lại chắc chắn sẽ xảy ra khi worker crash hoặc claim lại tin nhắn. Hãy đảm bảo tầng lưu trữ phía sau luôn sử dụng khóa idempotency hoặc ràng buộc duy nhất (unique constraint) của cơ sở dữ liệu.

Điểm Cốt Lõi

• Redis Pub/Sub phù hợp cho các tín hiệu tức thời không cần lưu trữ; Redis Streams cung cấp cấu trúc log bền vững kèm khả năng quản lý trạng thái consumer chặt chẽ.
• Danh sách Pending Entries List (PEL) theo dõi các tin nhắn đang được xử lý; chúng chỉ được gỡ bỏ khi consumer phát lệnh XACK tường minh.
• Sử dụng XAUTOCLAIM định kỳ để tự động thu hồi an toàn các thông điệp bị bỏ rơi do worker chết đột ngột.
• Luôn kết hợp bộ đếm số lần phân phối với Dead-Letter Queue để cách ly các thông điệp hỏng, bảo vệ đường ống xử lý không bị tắc nghẽn.