Series Redis Thực chiến | Bài 5: Redis Streams - Message Broker "Trùm cuối" với độ tin cậy tuyệt đối

Chào anh em! Ở Bài 4, chúng ta đã dùng Pub/Sub để giao tiếp giữa các module. Tuy nhiên, nếu một Service (Subscriber) bị crash đúng lúc tin nhắn được gửi đi, tin nhắn đó sẽ biến mất vĩnh viễn. Trong các hệ thống giao dịch vé (AFC) hay thanh toán, việc mất 1 tin nhắn đồng nghĩa với 1 giao dịch bị treo.

Redis Streams (ra mắt từ bản 5.0) là lời giải hoàn hảo. Nó biến Redis thành một Append-only log – nơi tin nhắn được lưu trữ bền vững trên RAM (và disk), cho phép Subscriber đọc lại bất cứ lúc nào, hỗ trợ cơ chế "Ack" (xác nhận đã xử lý xong) để đảm bảo không một tin nhắn nào bị bỏ sót.

1. Cơ chế vận hành: "Ghi nhật ký" siêu tốc

Thay vì gửi tin nhắn vào hư không, Redis Streams ghi tin nhắn vào một "dòng chảy" (Stream). Mỗi tin nhắn (Entry) sẽ có một ID duy nhất (ví dụ: 1719760000000-0).

• Producer: Thêm dữ liệu vào Stream bằng lệnh XADD.
• Consumer: Đọc dữ liệu từ Stream bằng lệnh XREAD hoặc XREADGROUP.
• Consumer Group: Đây là chìa khóa. Bạn có thể tạo một nhóm các Worker cùng đọc 1 Stream, Redis sẽ tự chia tải cho các Worker này (đảm bảo mỗi tin nhắn chỉ được xử lý bởi 1 Worker duy nhất).

2. Thực hành: Xử lý giao dịch vé tàu với Redis Streams

Bước 1: Producer (Bắn tin nhắn vào Stream) Khi có khách thanh toán vé xong, thay vì dùng Redis::publish, ta dùng Redis::xadd:

// Tạo tin nhắn vào Stream 'ticket_transactions'
Redis::xadd('ticket_transactions', '*', [
    'order_id' => 123,
    'status' => 'paid',
    'timestamp' => now()->toDateTimeString()
]);

Bước 2: Consumer Group (Xử lý bền vững) Đây là cách các hệ thống lớn xử lý việc đồng bộ dữ liệu vé. Chúng ta cần đăng ký một Group cho các Worker.

// Tạo Group 'processing_group' cho Stream 'ticket_transactions' (chạy 1 lần)
// '0' nghĩa là đọc từ đầu tiên
try {
    Redis::xgroup('CREATE', 'ticket_transactions', 'processing_group', '0', true);
} catch (\Exception $e) {
    // Group đã tồn tại, không sao cả
}

Bây giờ, Worker sẽ đọc và xác nhận (Ack) đã xử lý xong:

// Worker chạy liên tục
while (true) {
    // Đọc tin nhắn chưa được xử lý của consumer 'worker_1' trong group
    $messages = Redis::xreadgroup('GROUP', 'processing_group', 'worker_1', ['ticket_transactions' => '>'], 1, 5000);

    foreach ($messages['ticket_transactions'] as $id => $data) {
        try {
            // Xử lý nghiệp vụ: Đồng bộ dữ liệu xuống trạm AFC
            processTicketSync($data);

            // XÁC NHẬN: Báo cho Redis biết tin nhắn này đã xong, đừng gửi lại nữa
            Redis::xack('ticket_transactions', 'processing_group', $id);
        } catch (\Exception $e) {
            // Nếu lỗi, tin nhắn vẫn nằm đó, worker sau có thể lấy ra xử lý lại
        }
    }
}

3. Vì sao Redis Streams lại "hủy diệt" Pub/Sub?

1. Persistence (Bền vững): Dữ liệu được lưu trữ, không sợ mất khi Subscriber offline.
2. Acknowledgement (Ack): Chỉ khi Worker báo XACK, tin nhắn mới được coi là hoàn thành. Nếu Worker chết giữa chừng, tin nhắn đó sẽ chuyển vào "Pending Entries List", sẵn sàng cho một Worker khác nhảy vào xử lý tiếp.
3. Scalability: Bạn chỉ cần khởi chạy thêm 10 Worker nữa, Redis sẽ tự động cân bằng tải (Load balancing) tin nhắn cho 10 Worker đó.

Tổng kết Series Redis Thực chiến

Với 5 bài học, chúng ta đã đi từ tư duy "Redis không chỉ là Cache", dùng nó để khóa phân tán (Distributed Lock), chặn spam (Rate Limiting), phát thanh (Pub/Sub) và cuối cùng là làm hàng đợi tin nhắn bền vững (Streams).

Redis không chỉ là một món đồ chơi cho web nhanh, nó là một phần kiến trúc lõi của mọi hệ thống chịu tải lớn. Khi anh em đã nắm vững những thứ này, việc thiết kế một hệ thống backend chịu được triệu request/ngày không còn là chuyện viễn tưởng nữa.