# Bài 15: Redis Streams #### **Giới thiệu về Redis Streams:** * **Redis Streams là gì?** * Redis Streams là một cấu trúc dữ liệu mới trong Redis, được giới thiệu trong phiên bản 5.0. Streams cho phép lưu trữ và xử lý dữ liệu dạng luồng (streaming data) theo thứ tự thời gian, làm cho Redis trở thành một lựa chọn mạnh mẽ cho các ứng dụng cần xử lý dữ liệu thời gian thực. * **Lợi ích của Redis Streams:** * **Khả năng lưu trữ dữ liệu thời gian thực:** Streams cho phép ghi và đọc dữ liệu theo thứ tự thời gian, rất phù hợp cho các ứng dụng xử lý dữ liệu thời gian thực. * **Tính linh hoạt:** Redis Streams hỗ trợ nhiều tính năng mạnh mẽ như consumer groups, đảm bảo rằng dữ liệu được xử lý một cách hiệu quả và đáng tin cậy. * **Tích hợp dễ dàng:** Streams có thể được tích hợp vào các ứng dụng hiện có sử dụng Redis mà không cần thay đổi lớn trong kiến trúc. #### **Các lệnh cơ bản trong Redis Streams:** **1\. XADD:** * **Mô tả:** * Lệnh `XADD` dùng để thêm một entry (bản ghi) mới vào stream. * **Cú pháp:** ```plaintext XADD key [MAXLEN ~ count] * field value [field value ...] ``` * **Ví dụ:** ```plaintext XADD mystream * sensor-id 1234 temperature 19.8 ``` * Lệnh này thêm một entry mới vào stream `mystream` với các field `sensor-id` và `temperature`. **2\. XRANGE:** * **Mô tả:** * Lệnh `XRANGE` dùng để lấy các entry trong stream trong một phạm vi chỉ định. * **Cú pháp:** ```plaintext XRANGE key start end [COUNT count] ``` * **Ví dụ:** ```plaintext XRANGE mystream - + ``` * Lệnh này trả về tất cả các entry trong stream `mystream`. **3\. XREVRANGE:** * **Mô tả:** * Lệnh `XREVRANGE` dùng để lấy các entry trong stream trong một phạm vi chỉ định, theo thứ tự ngược lại. * **Cú pháp:** ```plaintext XREVRANGE key end start [COUNT count] ``` * **Ví dụ:** ```plaintext XREVRANGE mystream + - ``` * Lệnh này trả về tất cả các entry trong stream `mystream` theo thứ tự ngược lại. **4\. XREAD:** * **Mô tả:** * Lệnh `XREAD` dùng để đọc dữ liệu từ một hoặc nhiều streams. * **Cú pháp:** ```plaintext XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...] ``` * **Ví dụ:** ```plaintext XREAD STREAMS mystream 0 ``` * Lệnh này đọc tất cả các entry từ stream `mystream` bắt đầu từ đầu (ID = 0). **5\. XDEL:** * **Mô tả:** * Lệnh `XDEL` dùng để xóa một hoặc nhiều entry khỏi stream. * **Cú pháp:** ```plaintext XDEL key ID [ID ...] ``` * **Ví dụ:** ```plaintext XDEL mystream 1526985052478-0 ``` * Lệnh này xóa entry có ID `1526985052478-0` khỏi stream `mystream`. #### **Consumer Groups trong Redis Streams:** **1\. XGROUP CREATE:** * **Mô tả:** * Lệnh `XGROUP CREATE` dùng để tạo một consumer group cho stream. * **Cú pháp:** ```plaintext XGROUP CREATE key groupname id-or-$ ``` * **Ví dụ:** ```plaintext XGROUP CREATE mystream mygroup $ ``` * Lệnh này tạo một consumer group tên `mygroup` cho stream `mystream`, bắt đầu từ entry mới nhất (`$`). **2\. XREADGROUP:** * **Mô tả:** * Lệnh `XREADGROUP` dùng để đọc dữ liệu từ stream như một phần của consumer group. * **Cú pháp:** ```plaintext XREADGROUP GROUP groupname consumername [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...] ``` * **Ví dụ:** ```plaintext XREADGROUP GROUP mygroup consumer1 STREAMS mystream > ``` * Lệnh này đọc các entry mới từ stream `mystream` như một phần của consumer `consumer1` trong group `mygroup`. **3\. XACK:** * **Mô tả:** * Lệnh `XACK` dùng để xác nhận rằng một hoặc nhiều entry đã được xử lý bởi consumer group. * **Cú pháp:** ```plaintext XACK key groupname ID [ID ...] ``` * **Ví dụ:** ```plaintext XACK mystream mygroup 1526985052478-0 ``` * Lệnh này xác nhận rằng entry có ID `1526985052478-0` trong stream `mystream` đã được xử lý bởi group `mygroup`. **4\. XPENDING:** * **Mô tả:** * Lệnh `XPENDING` dùng để xem các entry đang chờ xử lý (pending) trong consumer group. * **Cú pháp:** ```plaintext XPENDING key groupname [start end count] [consumer] ``` * **Ví dụ:** ```plaintext XPENDING mystream mygroup ``` * Lệnh này liệt kê tất cả các entry đang chờ xử lý trong group `mygroup` của stream `mystream`. #### **Ví dụ thực tiễn:** **Xây dựng hệ thống stream processing đơn giản:** **Scenario:** Bạn cần xây dựng một hệ thống thu thập và xử lý dữ liệu cảm biến nhiệt độ thời gian thực. **Bước 1: Tạo và thêm dữ liệu vào stream:** ```plaintext XADD sensor_data * sensor-id 1 temperature 22.5 XADD sensor_data * sensor-id 2 temperature 23.1 XADD sensor_data * sensor-id 1 temperature 21.8 ``` **Bước 2: Tạo consumer group để xử lý dữ liệu:** ```plaintext XGROUP CREATE sensor_data sensor_group $ ``` **Bước 3: Consumer đọc và xử lý dữ liệu từ stream:** ```plaintext XREADGROUP GROUP sensor_group consumer1 STREAMS sensor_data > ``` **Bước 4: Xác nhận dữ liệu đã được xử lý:** ```plaintext XACK sensor_data sensor_group 1526985052478-0 ``` **Bước 5: Kiểm tra các entry đang chờ xử lý:** ```plaintext XPENDING sensor_data sensor_group ``` #### **Các tính năng bổ sung của Redis Streams:** **1\. ID Management:** * **Mô tả:** * Redis Streams sử dụng một định dạng ID đặc biệt cho mỗi entry, bao gồm một timestamp và một sequence number để đảm bảo tính duy nhất. **2\. Trimming:** * **Mô tả:** * Streams hỗ trợ tính năng trimming để giới hạn số lượng entry, giúp kiểm soát kích thước của stream. **3\. Persistence:** * **Mô tả:** * Dữ liệu trong Redis Streams được lưu trữ bền vững và có thể được cấu hình để ghi vào đĩa (disk persistence). **4\. Stream Length:** * **Mô tả:** * Lệnh `XLEN` dùng để lấy số lượng entry hiện có trong stream. * **Cú pháp:** ```plaintext XLEN key ``` * **Ví dụ:** ```plaintext XLEN sensor_data ``` #### **Câu hỏi củng cố kiến thức:** 1. **Redis Streams là gì và lợi ích của nó là gì?** * Redis Streams là một cấu trúc dữ liệu trong Redis cho phép lưu trữ và xử lý dữ liệu dạng luồng theo thứ tự thời gian, giúp xử lý dữ liệu thời gian thực hiệu quả. 2. **Lệnh nào dùng để thêm một entry mới vào stream?** * `XADD`. 3. **Làm thế nào để tạo một consumer group cho stream?** * Sử dụng lệnh `XGROUP CREATE`. 4. **Lệnh nào dùng để đọc dữ liệu từ stream như một phần của consumer group?** * `XREADGROUP`. 5. **Làm thế nào để xác nhận rằng một entry đã được xử lý bởi consumer group?** * Sử dụng lệnh `XACK`. ### Kết luận: * Redis Streams cung cấp một giải pháp mạnh mẽ và linh hoạt để xử lý dữ liệu thời gian thực, giúp các ứng dụng có thể xử lý và phân tích dữ liệu một cách nhanh chóng và hiệu quả. Bằng cách sử dụng các lệnh và tính năng của Redis Streams, bạn có thể dễ dàng xây dựng các hệ thống stream processing mạnh mẽ và đáng tin cậy, phù hợp với nhiều loại ứng dụng và trường hợp sử dụng khác nhau. Việc hiểu và áp dụng các lệnh của Redis Streams sẽ giúp bạn khai thác tối đa khả năng của Redis trong việc xử lý dữ liệu thời gian thực.