1. Chiến lược mã hóa kép của Redis Set

Trong Redis, Set là một tập hợp các chuỗi không trùng lặp và không có thứ tự. Để tối ưu hóa giữa hiệu suất truy cập và dung lượng bộ nhớ, Redis sử dụng hai cấu trúc lưu trữ khác nhau cho Set: intset (tập hợp số nguyên) và hashtable (bảng băm). Việc chuyển đổi giữa hai cấu trúc này diễn ra tự động dựa trên đặc điểm của dữ liệu.

Một đối tượng Set sẽ sử dụng cấu trúc intset nếu đáp ứng đồng thời hai điều kiện sau:

• Tất cả các phần tử được thêm vào đều là số nguyên.
• Số lượng phần tử không vượt quá cấu hình set-max-intset-entries (mặc định là 512).

Nếu một trong hai điều kiện trên bị vi phạm (ví dụ: thêm một chuỗi không phải số hoặc số lượng phần tử vượt ngưỡng), Redis sẽ chuyển đổi Set đó sang cấu trúc hashtable.

2. Chi tiết cấu trúc dữ liệu Intset

intset là một cấu trúc mảng đặc biệt, được thiết kế để lưu trữ các số nguyên một cách cô đọng nhất. Điểm đặc biệt của nó là khả năng tự nâng cấp kích thước lưu trữ (encoding) dựa trên giá trị lớn nhất hiện có trong tập hợp.

Cấu trúc C của Intset:

typedef struct intset {
    uint32_t mode;      // Chế độ mã hóa: INTSET_ENC_INT16, INT32 hoặc INT64
    uint32_t count;     // Tổng số phần tử hiện có
    int8_t data[];      // Mảng chứa các phần tử thực tế
} intset;

Cơ chế nâng cấp (Upgrade)

Khi thêm một số nguyên mới có kích thước lớn hơn khả năng lưu trữ hiện tại (ví dụ: đang ở 16-bit nhưng thêm một số cần 32-bit), Redis sẽ thực hiện quy trình nâng cấp:

1. Cấp phát lại bộ nhớ cho mảng theo kích thước mới.
2. Dịch chuyển các phần tử cũ sang vị trí mới tương ứng với độ dài bit mới.
3. Thêm phần tử mới vào cuối hoặc đầu mảng (vì phần tử gây ra việc nâng cấp luôn là lớn nhất hoặc nhỏ nhất).

Lưu ý: intset không hỗ trợ hạ cấp (downgrade) để tránh việc cấp phát lại bộ nhớ liên tục, gây ảnh hưởng hiệu suất.

3. Tìm kiếm và Thao tác trên Intset

Vì intset lưu trữ dữ liệu dưới dạng mảng đã được sắp xếp, Redis sử dụng thuật toán tìm kiếm nhị phân để kiểm tra sự tồn tại của phần tử, mang lại độ phức tạp thời gian O(log N).

Mô phỏng logic tìm kiếm trong Intset:

static uint8_t findPosition(intset *is, int64_t target, uint32_t *pos) {
    int low = 0, high = intsetLen(is) - 1;
    int64_t mid_val;

    if (intsetLen(is) == 0) {
        if (pos) *pos = 0;
        return 0;
    }

    // Kiểm tra nhanh các biên
    if (target > getVal(is, high)) {
        if (pos) *pos = intsetLen(is);
        return 0;
    } else if (target < getVal(is, 0)) {
        if (pos) *pos = 0;
        return 0;
    }

    while(high >= low) {
        int mid = (low + high) / 2;
        mid_val = getVal(is, mid);
        if (target > mid_val) {
            low = mid + 1;
        } else if (target < mid_val) {
            high = mid - 1;
        } else {
            if (pos) *pos = mid;
            return 1;
        }
    }

    if (pos) *pos = low;
    return 0;
}

4. So sánh hiệu quả bộ nhớ và hiệu suất

Bảng dưới đây so sánh sự khác biệt khi lưu trữ 500 số nguyên nhỏ giữa hai loại mã hóa:

5. Ứng dụng thực tế và Mã nguồn minh họa

Hệ thống gắn nhãn (Tagging System)

Sử dụng Set để quản lý các nhãn gắn với sản phẩm và ngược lại.

import redis

class TagManager:
    def __init__(self, host='localhost', port=6379):
        self.client = redis.StrictRedis(host=host, port=port, decode_responses=True)

    def attach_tags(self, item_id, tag_list):
        pipe = self.client.pipeline()
        for tag_name in tag_list:
            # Lưu sản phẩm vào tập hợp của tag
            pipe.sadd(f"tag_members:{tag_name}", item_id)
            # Lưu tag vào tập hợp của sản phẩm
            pipe.sadd(f"item_tags:{item_id}", tag_name)
        pipe.execute()

    def find_items_by_tags(self, tags, logic='AND'):
        keys = [f"tag_members:{t}" for t in tags]
        if logic == 'AND':
            return self.client.sinter(keys)
        return self.client.sunion(keys)

Theo dõi người dùng duy nhất (Unique User Tracking)

Thống kê số lượng người dùng truy cập trong ngày, đảm bảo không trùng lặp.

class ActivityTracker:
    def __init__(self, r_client):
        self.r = r_client

    def log_activity(self, user_id, category):
        from datetime import datetime
        today = datetime.now().strftime("%Y%m%d")
        key = f"active_users:{category}:{today}"
        self.r.sadd(key, user_id)

    def get_unique_count(self, category, date_str):
        return self.r.scard(f"active_users:{category}:{date_str}")

Hệ thống quay số trúng thưởng (Raffle System)

Tận dụng lệnh SPOP hoặc SRANDMEMBER để chọn người thắng cuộc một cách ngẫu nhiên và công bằng.

class RaffleSystem:
    def __init__(self, r_conn):
        self.redis = r_conn

    def enter_raffle(self, raffle_id, user_id):
        return self.redis.sadd(f"raffle:{raffle_id}:pool", user_id)

    def pick_winners(self, raffle_id, quantity):
        pool_key = f"raffle:{raffle_id}:pool"
        winner_key = f"raffle:{raffle_id}:winners"
        
        # Lấy ngẫu nhiên và xóa khỏi danh sách chờ
        winners = self.redis.spop(pool_key, quantity)
        if winners:
            self.redis.sadd(winner_key, *winners)
        return winners

6. Lưu ý về hiệu năng khi làm việc với Set

1. Kích thước tập hợp: Các phép toán tập hợp như SINTER, SUNION, SDIFF có độ phức tạp cao. Với các tập hợp lớn, nên thực hiện trên các máy phụ (slave) hoặc sử dụng SINTERSTORE để lưu kết quả tạm thời.
2. Phân đoạn dữ liệu: Nếu một Set có hàng triệu phần tử, nó có thể gây ra hiện tượng "Big Key". Trong trường hợp này, nên chia nhỏ Set theo một thuật toán băm (sharding).
3. Tối ưu Intset: Nếu ứng dụng của bạn chủ yếu lưu các ID dạng số, hãy cân nhắc tăng set-max-intset-entries để tận dụng tối đa khả năng tiết kiệm bộ nhớ của intset, nhưng không nên để quá cao (trên 1024) vì thao tác O(N) trên mảng lớn sẽ làm chậm CPU.