Bài 9: Redis Streams - Xây dựng Message Queue bền vững với Consumer Groups

Mục tiêu bài học:

Hiểu sự vượt trội của Redis Streams so với Pub/Sub (không mất dữ liệu) và List (hỗ trợ phân phối tải phức tạp).
Triển khai Consumer Groups để chia sẻ tải giữa nhiều service con (Load Balancing).
Đảm bảo tính toàn vẹn dữ liệu với cơ chế ACK (Acknowledge) theo chuẩn At-least-once delivery.

1. Sự tiến hóa: Tại sao lại là Redis Streams?

Nếu Pub/Sub giống như đài phát thanh (ai không nghe lúc đó sẽ mất tin) và List giống như một hàng đợi xử lý tuần tự (một tin nhắn chỉ được một người lấy ra), thì Redis Streams (ra mắt từ Redis 5.0) là sự kết hợp hoàn hảo của cả hai và mang dáng dấp của Apache Kafka.

Bền vững (Persistence): Dữ liệu gửi vào Stream được lưu trữ lại trên ổ cứng (tùy cấu hình Redis). Consumer offline khi bật lại vẫn có thể đọc tiếp từ điểm dừng.
Định danh độc nhất (Auto ID): Mỗi tin nhắn có một ID duy nhất theo thời gian (ví dụ: 1696085429123-0), cho phép truy xuất chính xác lịch sử.
Consumer Groups: Nhóm nhiều Consumer lại với nhau để cùng "chia sẻ" việc đọc một Stream. Một tin nhắn trong Stream chỉ được giao cho một Consumer trong Group đó xử lý, giúp tăng tốc độ (Scaling).

2. Producer: Đẩy dữ liệu vào Stream (XADD)

Lệnh XADD dùng để thêm một entry (chứa các cặp field-value) vào Stream.

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

var ctx = context.Background()

func produceToStream(rdb *redis.Client, streamName string) {
	for i := 1; i <= 3; i++ {
		// Tham số truyền vào là struct XAddArgs
		err := rdb.XAdd(ctx, &redis.XAddArgs{
			Stream: streamName,
			// ID: "*", // Dấu * để Redis tự động tạo ID dựa trên timestamp (mặc định)
			Values: map[string]interface{}{
				"order_id": fmt.Sprintf("ORD_%d", i),
				"status":   "pending",
				"amount":   100 * i,
			},
		}).Err()

		if err != nil {
			fmt.Println("Lỗi đẩy dữ liệu vào Stream:", err)
		} else {
			fmt.Printf("[Producer] Đã gửi đơn hàng ORD_%d\n", i)
		}
		time.Sleep(500 * time.Millisecond)
	}
}

3. Khởi tạo Consumer Group (XGROUP CREATE)

Trước khi đọc dữ liệu theo nhóm, bạn cần định nghĩa Consumer Group. Việc này chỉ thực hiện 1 lần duy nhất cho mỗi Group.

func createConsumerGroup(rdb *redis.Client, streamName, groupName string) {
	// XGroupCreateMkStream: Tạo Group, nếu Stream chưa tồn tại thì tạo luôn Stream (MkStream)
	// Ký hiệu "$" nghĩa là Group này chỉ quan tâm tới những tin nhắn MỚI phát sinh từ bây giờ.
	// Ký hiệu "0" nếu muốn Group đọc lại từ đầu lịch sử của Stream.
	err := rdb.XGroupCreateMkStream(ctx, streamName, groupName, "$").Err()
	
	if err != nil {
		// Hàm này sẽ trả về lỗi "BUSYGROUP Consumer Group name already exists" nếu group đã tạo
		// Trong thực tế chúng ta có thể bỏ qua lỗi này
		if err.Error() != "BUSYGROUP Consumer Group name already exists" {
			fmt.Println("Lỗi tạo Consumer Group:", err)
		}
	} else {
		fmt.Printf("Đã khởi tạo Consumer Group '%s' thành công.\n", groupName)
	}
}

4. Consumer: Lắng nghe và Xác nhận (XREADGROUP & XACK)

Đây là vòng đời hoàn chỉnh của một Worker trong hệ thống Microservices: Lắng nghe việc mới $\rightarrow$ Xử lý logic $\rightarrow$ Đánh dấu hoàn thành (ACK).

Nếu Worker bị sập lúc đang xử lý (chưa ACK), tin nhắn đó vẫn nằm trong danh sách Pending (PEL - Pending Entries List) của Group và có thể được Worker khác claim (nhận lại) để xử lý tiếp, đảm bảo không bao giờ rớt đơn hàng.

func consumeFromStream(rdb *redis.Client, streamName, groupName, consumerName string) {
	fmt.Printf("[%s] Bắt đầu lắng nghe Stream...\n", consumerName)

	for {
		// 1. Đọc tin nhắn qua Group
		streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    groupName,
			Consumer: consumerName,
			Streams:  []string{streamName, ">"}, // ">" nghĩa là lấy tin nhắn chưa từng được giao cho ai trong Group
			Count:    1,                         // Lấy 1 tin mỗi lần
			Block:    0,                         // 0 = Block vô hạn cho đến khi có tin mới
		}).Result()

		if err != nil {
			if err == redis.Nil {
				continue // Timeout, quay lại vòng lặp
			}
			fmt.Println("Lỗi khi đọc Stream:", err)
			time.Sleep(1 * time.Second)
			continue
		}

		// 2. Xử lý tin nhắn
		// Cấu trúc trả về khá sâu: Array của Streams -> Array của Messages -> Message cụ thể
		for _, stream := range streams {
			for _, msg := range stream.Messages {
				orderID := msg.Values["order_id"]
				fmt.Printf("[%s] Đang xử lý tin nhắn ID: %s - Data: %v\n", 
					consumerName, msg.ID, orderID)

				// Giả lập thời gian xử lý nghiệp vụ (lưu DB, gọi API ngoài...)
				time.Sleep(1 * time.Second)

				// 3. Đánh dấu hoàn thành (ACK)
				// Nếu không có dòng này, tin nhắn sẽ kẹt ở trạng thái Pending mãi mãi
				err := rdb.XAck(ctx, streamName, groupName, msg.ID).Err()
				if err != nil {
					fmt.Printf("[%s] Lỗi ACK tin nhắn %s: %v\n", consumerName, msg.ID, err)
				} else {
					fmt.Printf("[%s] Đã hoàn thành và ACK %s\n", consumerName, orderID)
				}
			}
		}
	}
}

Cách chạy thử nghiệm:

Bạn có thể kết hợp các hàm trên trong main() bằng cách chạy nhiều Goroutine đóng vai trò các Consumer khác nhau (ví dụ: worker_1, worker_2) trong cùng một group_orders. Khi đó, Redis sẽ tự động chia đều các order_id cho 2 worker, không có order nào bị xử lý trùng 2 lần.

Backend

1. Sự tiến hóa: Tại sao lại là Redis Streams?

2. Producer: Đẩy dữ liệu vào Stream (XADD)

3. Khởi tạo Consumer Group (XGROUP CREATE)

4. Consumer: Lắng nghe và Xác nhận (XREADGROUP & XACK)

Mục lục