Bài 8: Pub/Sub System - Giao tiếp thời gian thực và Bắn sự kiện

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

• Nắm bắt mô hình Publisher (Người phát) và Subscriber (Người đăng ký) trong Redis.

• Sử dụng go-redis để bắn và lắng nghe các sự kiện theo thời gian thực.

• Phân biệt đặc tính "Fire-and-Forget" của Pub/Sub so với Message Queue truyền thống.

1. Mô hình Pub/Sub trong Redis là gì?

Khác với các cấu trúc dữ liệu lưu trữ (như String, List, Hash), Pub/Sub không lưu trữ bất kỳ dữ liệu nào. Nó hoạt động như một hệ thống trạm phát thanh:

• Publisher phát một thông điệp (Payload) vào một kênh (Channel).

• Bất kỳ Subscriber nào đang dò đúng kênh đó vào thời điểm phát sẽ nhận được thông điệp.

• Nếu Subscriber bị ngắt kết nối tạm thời lúc thông điệp được phát, thông điệp đó sẽ mất vĩnh viễn đối với họ (Fire-and-Forget).

Ứng dụng thực tế:

• Hệ thống Chat (Phòng chat, livestream comments).

• Bắn thông báo (Push Notifications) thời gian thực tới Websocket server.

• Phân phối sự kiện (Event Broadcasting) giữa các microservices (ví dụ: Service A cập nhật sản phẩm, bắn event qua Pub/Sub để các Service B, C, D xóa local cache).

2. Triển khai Subscriber (Lắng nghe sự kiện)

Để lắng nghe, chúng ta gọi hàm Subscribe. Lệnh này trả về một đối tượng PubSub, từ đó có thể trích xuất ra một Go Channel (<-chan *redis.Message) để nhận dữ liệu liên tục bằng vòng lặp for.

Go

package main

import (
	"context"
	"fmt"
	"time"

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

var ctx = context.Background()

func startSubscriber(rdb *redis.Client, channelName string) {
	// 1. Đăng ký vào một kênh
	sub := rdb.Subscribe(ctx, channelName)
	
	// QUAN TRỌNG: Phải đóng PubSub khi không dùng nữa để giải phóng kết nối mạng
	defer sub.Close()

	// 2. Chuyển đổi thành Go channel để nhận message dễ dàng
	ch := sub.Channel()

	fmt.Printf("[Subscriber] Đang lắng nghe kênh '%s'...\n", channelName)
	
	// 3. Vòng lặp này sẽ block và chờ tin nhắn đến
	for msg := range ch {
		fmt.Printf("[Subscriber] Nhận tin từ '%s': %s\n", msg.Channel, msg.Payload)
	}
}

3. Triển khai Publisher (Phát sự kiện)

Việc phát sự kiện rất đơn giản và diễn ra ngay lập tức. Hàm Publish trả về một số nguyên thể hiện số lượng client đang kết nối và đã nhận được thông điệp này.

Go

func startPublisher(rdb *redis.Client, channelName string) {
	messages := []string{
		"Hệ thống chuẩn bị bảo trì trong 5 phút",
		"Có đơn hàng mới #10029",
		"User 999 vừa đăng nhập",
	}

	for _, msg := range messages {
		// Phát tin nhắn vào kênh
		receivers, err := rdb.Publish(ctx, channelName, msg).Result()
		if err != nil {
			fmt.Println("[Publisher] Lỗi khi phát tin:", err)
			continue
		}
		
		fmt.Printf("[Publisher] Gửi: '%s' -> (Có %d client nhận được)\n", msg, receivers)
		time.Sleep(1 * time.Second) // Giả lập độ trễ giữa các lần phát
	}
}

4. Kết hợp chạy thực tế (Chạy đồng thời)

Sử dụng Goroutines để khởi chạy Subscriber chạy ngầm, sau đó dùng Thread chính để đóng vai trò Publisher.

Go

func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})
	defer rdb.Close()

	channel := "system_events"

	// Bật 2 Subscriber chạy nền (Goroutine)
	// Mỗi người đăng ký sẽ nhận được một bản sao của cùng một thông điệp
	go startSubscriber(rdb, channel)
	go startSubscriber(rdb, channel)

	// Đợi một chút để các Subscriber kịp thiết lập kết nối mạng
	time.Sleep(500 * time.Millisecond)

	// Main thread đóng vai trò Publisher
	startPublisher(rdb, channel)

	// Đợi các goroutine in log xong
	time.Sleep(1 * time.Second)
}

5. Lắng nghe nhiều kênh với Pattern Matching (PSUBSCRIBE)

Nếu hệ thống của bạn có nhiều kênh (ví dụ: orders:create, orders:update, orders:delete), bạn có thể lắng nghe tất cả thông qua thẻ wildcard * bằng lệnh PSubscribe.

Go

func pSubscribeExample(rdb *redis.Client) {
	// Lắng nghe tất cả các kênh bắt đầu bằng "orders:"
	sub := rdb.PSubscribe(ctx, "orders:*")
	defer sub.Close()

	ch := sub.Channel()
	for msg := range ch {
		// msg.Pattern = "orders:*" (Mẫu dùng để đăng ký)
		// msg.Channel = Kênh thực tế phát ra (vd: "orders:create")
		fmt.Printf("[PSub] Pattern: %s | Channel: %s | Payload: %s\n", 
			msg.Pattern, msg.Channel, msg.Payload)
	}
}