Redis 設計與實現 9：五大數據類型之集合

集合對象的編碼有兩種：intset 和 hashtable

編碼一：intset

intset 的結構

整數集合 intset 是集合底層的實現之一，從名字就能夠看出，這是專門爲整數提供的集合類型。
其結構定義以下，在 intset.h：

typedef struct intset {
    // 編碼方式
    uint32_t encoding;
    // 集合包含的元素數量
    uint32_t length;
    // 保存元素的數組
    int8_t contents[];
} intset;

• contents 中的元素，按照從小到大排序，而且不存在重複項。雖然元素定義是 int8_t 類型，但實際上，contents 存的元素類型取決於 encoding
• encoding 有幾個類型，定義在 intset.c：

#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))

encoding	類型	字節
INTSET_ENC_INT16	int16_t	2
INTSET_ENC_INT32	int32_t	4
INTSET_ENC_INT64	int64_t	8

下圖展現了包含了 一、二、3 三個整數元素的集合結構：

常見操做源碼分析

源碼在 intset.c 中

1. 建立空集合

建立一個空的 intset，一開始的編碼是最小的 INTSET_ENC_INT16

intset *intsetNew(void) {
    intset *is = zmalloc(sizeof(intset));
    is->encoding = intrev32ifbe(INTSET_ENC_INT16);
    is->length = 0;
    return is;
}

2. 搜索

由於集合中的整數存的是有序的，因此查找是用二分查找，時間複雜度 \(O(nlogn)\)

uint8_t intsetFind(intset *is, int64_t value) {
    uint8_t valenc = _intsetValueEncoding(value);
    // 若是 value 的編碼大於集合的編碼，那確定是不存在的
    // intsetSearch 是更底層的搜索，實現源碼在下面，是個二分查找
    return valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,NULL);
}

// 集合搜索，是二分查找。
// 若是找到了，返回1，而且把位置設置到 pos 變量中
// 若是找不到，返回0，能夠插入值的位置設置到 pos 變量中
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
    int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;
    int64_t cur = -1;

    // 數組判空
    if (intrev32ifbe(is->length) == 0) {
        if (pos) *pos = 0;
        return 0;
    } else {
        // 看是否比最大的大或者比最小的小，這種狀況也直接返回不存在
        if (value > _intsetGet(is,max)) {
            if (pos) *pos = intrev32ifbe(is->length);
            return 0;
        } else if (value < _intsetGet(is,0)) {
            if (pos) *pos = 0;
            return 0;
        }
    }

    // 二分查找
    while(max >= min) {
        mid = ((unsigned int)min + (unsigned int)max) >> 1;
        cur = _intsetGet(is,mid);
        if (value > cur) {
            min = mid+1;
        } else if (value < cur) {
            max = mid-1;
        } else {
            break;
        }
    }

    if (value == cur) {
        if (pos) *pos = mid;
        return 1;
    } else {
        if (pos) *pos = min;
        return 0;
    }
}

3. 指定位置獲取

// 若是獲取獲得，返回1，找到的值設置進 value 變量
// 若是獲取不到，返回 0
uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value) {
    if (pos < intrev32ifbe(is->length)) {
        *value = _intsetGet(is,pos);
        return 1;
    }
    // 位置若是大於長度，確定就獲取不到的
    return 0;
}
static int64_t _intsetGet(intset *is, int pos) {
    // 根據編碼獲取
    return _intsetGetEncoded(is,pos,intrev32ifbe(is->encoding));
}
static int64_t _intsetGetEncoded(intset *is, int pos, uint8_t enc) {
    int64_t v64;
   	// ...

    // 根據編碼的長度，從對應的位置後拷貝對應的字節返回
    if (enc == INTSET_ENC_INT64) {
        memcpy(&v64,((int64_t*)is->contents)+pos,sizeof(v64));
        memrev64ifbe(&v64);
        return v64;
    } else if (enc == INTSET_ENC_INT32) {
        // ...
        return v32;
    } else {
        // ...
    }
}

4. 插入

插入的步驟以下：

1. 檢查若是插入的元素的編碼大於集合編碼，進行升級並插入
2. 若是不須要升級，檢查元素是否存在，若是存在，則直接返回
3. 若是元素不存在，則擴容，在元素對應的位置插入值(它後面的元素則都日後挪)

intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
    // 插入的元素的編碼
    uint8_t valenc = _intsetValueEncoding(value);
    uint32_t pos;
    if (success) *success = 1;

    // 若是插入的元素的編碼比當前集合的編碼大，須要進行升級
    if (valenc > intrev32ifbe(is->encoding)) {
        return intsetUpgradeAndAdd(is,value);
    } else {
        // 先查找看元素已存在，若是存在，則直接返回
        if (intsetSearch(is,value,&pos)) {
            if (success) *success = 0;
            return is;
        }
		
		// 擴容
        is = intsetResize(is,intrev32ifbe(is->length)+1);
        // 將 pos 後的內存塊向後挪動一個位置，給新值騰空間
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
    }

    // 把新值設置進 pos 位置上
    _intsetSet(is,pos,value);
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) {
    void *src, *dst;
    uint32_t bytes = intrev32ifbe(is->length)-from;
    uint32_t encoding = intrev32ifbe(is->encoding);

    if (encoding == INTSET_ENC_INT64) {
        src = (int64_t*)is->contents+from;
        dst = (int64_t*)is->contents+to;
        bytes *= sizeof(int64_t);
    } else if (encoding == INTSET_ENC_INT32) {
        // ...
    } else {
        // ...
    }
    memmove(dst,src,bytes);
}

5. 升級

當 intset 插入元素的時候，會先檢測元素的長度，判斷元素應該屬於什麼編碼(encoding)。
若是當前元素的編碼，大於 intset 的編碼(整個集合最長的編碼)，集合將進行升級後，才添加元素。

升級整數集合並添加新元素共分爲 3 步進行：

1. 根據新元素的編碼，擴展整數集合底層數組的空間大小，併爲新元素分配空間。
2. 將底層數組現有的全部元素都轉換成與新元素相同的類型，並將類型轉換後的元素放置到正確的位上，並且在放置元素的過程當中，須要繼續維持底層數組的有序性質不變。
3. 將新元素添加到底層數組裏面。

// 升級並插入新值
static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
    // 當前編碼
    uint8_t curenc = intrev32ifbe(is->encoding);
    // 新的編碼
    uint8_t newenc = _intsetValueEncoding(value);
    // 當前元素個數
    int length = intrev32ifbe(is->length);
    // value 的編碼比其餘的都大，那麼這個 value 不是最大值就是最小值。
    // 若是是最大值就放在數組最後，最小值就放在數組最前面
    int prepend = value < 0 ? 1 : 0;

    // 設置 encoding 屬性爲新編碼
    is->encoding = intrev32ifbe(newenc);
    // 根據新編碼給擴展集合須要的空間，實現源碼在下面
    is = intsetResize(is,intrev32ifbe(is->length)+1);

    // 從尾到頭依次遍歷挪動原來的值。爲何不從頭至尾呢？由於數組是同一個，從頭至尾會覆蓋原來的值
    while(length--)
        // _intsetGetEncoded(is,length,curenc) 表示根據編碼和位置獲取值
        // prepend 爲了確保若是 value 是最小的值，那麼前面會留一個空位置
        _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));

    if (prepend)
    	// 當 value 是最小值時，放在第一個空位
        _intsetSet(is,0,value);
    else
        // 當 value 是最大值，放在最後一個位置
        _intsetSet(is,intrev32ifbe(is->length),value);
    // 長度加 1
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

// 整數集合從新分配內存
static intset *intsetResize(intset *is, uint32_t len) {
    // 根據編碼算出集合須要的空間
    uint32_t size = len*intrev32ifbe(is->encoding);
    // 分配內存
    is = zrealloc(is,sizeof(intset)+size);
    return is;
}

6. 降級

並無降級

7. 刪除

刪除的步驟以下：

1. 找到值的位置 pos
2. 把 pos 後面的元素向前挪，覆蓋掉 pos 上的元素
3. 縮容：長度減一

intset *intsetRemove(intset *is, int64_t value, int *success) {
    uint8_t valenc = _intsetValueEncoding(value);
    uint32_t pos;
    if (success) *success = 0;

    // 查找值的位置
    if (valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,&pos)) {
        uint32_t len = intrev32ifbe(is->length);
        if (success) *success = 1;
        // 把刪除位置後面的元素都挪到前面來，直接覆蓋掉 pos 的元素
        if (pos < (len-1)) intsetMoveTail(is,pos+1,pos);
        // 再縮容
        is = intsetResize(is,len-1);
        is->length = intrev32ifbe(len-1);
    }
    return is;
}

---

編碼二：hashtable

hashtable 編碼用的是字典 dict 做爲底層實現，關於 dict，具體的前文 Redis 設計與實現 4：字典 dict 已經寫了，包括了 dict 基本操做的源碼解讀。

下圖展現了包含 "a"、"b"、"c"、"d" 四個元素的集合結構：

編碼的轉換

當集合對象知足如下兩個條件時，採用 intset 編碼：

1. 全部元素都是整數
2. 元素數量不超過512個(用經過 set-max-intset-entries 配置項配置)

不能同時知足以上兩個條件，則採用 tablehash 編碼。