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:

    XADD key [MAXLEN ~ count] * field value [field value ...]
  

• Ví dụ:

    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:

    XRANGE key start end [COUNT count]
  

• Ví dụ:

    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:

    XREVRANGE key end start [COUNT count]
  

• Ví dụ:

    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:

    XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
  

• Ví dụ:

    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:

    XDEL key ID [ID ...]
  

• Ví dụ:

    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:

    XGROUP CREATE key groupname id-or-$
  

• Ví dụ:

    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:

    XREADGROUP GROUP groupname consumername [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
  

• Ví dụ:

    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:

    XACK key groupname ID [ID ...]
  

• Ví dụ:

    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:

    XPENDING key groupname [start end count] [consumer]
  

• Ví dụ:

    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:

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:

XGROUP CREATE sensor_data sensor_group $

Bước 3: Consumer đọc và xử lý dữ liệu từ stream:

XREADGROUP GROUP sensor_group consumer1 STREAMS sensor_data >

Bước 4: Xác nhận dữ liệu đã được xử lý:

XACK sensor_data sensor_group 1526985052478-0

Bước 5: Kiểm tra các entry đang chờ xử lý:

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:

      XLEN key
    

  • Ví dụ:

      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.