一、集合概述

? ? ? ? 對于集合，STL 的 set 相信大家都不陌生，它的底層實現(xiàn)是紅黑樹。無論插入、刪除、查找都是 O(log n) 的時間復雜度。當然，如果用哈希表來實現(xiàn)集合，插入、刪除、查找都可以達到 O(1)。那么為什么集合要用紅黑樹和沒有用哈希表呢？我想，最大的可能是基于集合自身的特性，集合有它特有的操作：求交、求并、求差。這三個操作對于哈希表來說都是 O(n) 的。基于這一點，相比無序的哈希表來說，采用有序的紅黑樹會更加合適。

二、Redis 整數(shù)集合（intset）

? ? ? ? 今天要講的整數(shù)集合，又稱為 intset，是 Redis 特有的數(shù)據(jù)結(jié)構(gòu)。它的實現(xiàn)既不是紅黑樹，也不是哈希表。就是簡單的數(shù)組加上內(nèi)存編碼。當存儲元素較少（ 元素個數(shù)上限定義在server.h 的 OBJ_SET_MAX_INTSET_ENTRIES 宏定義值為512）且均為整型時，才會使用到整數(shù)集合。它的查找是 O(log n) 的，插入和刪除都是 O(n) 的。但是由于存儲元素相對較少的時候，O(log n) 和?O(n) 差距不是很大，但是用 Redis 的這種整數(shù)集合，相比紅黑樹和哈希表來說，可以大大減少內(nèi)存。
? ? ? ? 所以，Redis 的 整數(shù)集合 intset 的存在主要還是為了節(jié)省內(nèi)存。

1、intset 結(jié)構(gòu)定義

? ? ? ? intset 結(jié)構(gòu)定義在 intset.h 中：

#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))
 
typedef struct intset {
    uint32_t encoding;      /* a */
    uint32_t length;        /* b */
    int8_t contents[];      /* c */
} intset;

? ? ? ? a) encoding 指定了編碼方式，總共有 INTSET_ENC_INT16、INTSET_ENC_INT32、INTSET_ENC_INT64 三種。從宏定義可以看出，這三個值分別為 2、4、8。從字面意思可以看出三者能表示的范圍是 16位整數(shù)、32位整數(shù) 以及 64位整數(shù)。
? ? ? ? b) length 存儲了整數(shù)集合的元素個數(shù)。
? ? ? ? c) contents 為整數(shù)集合的柔性數(shù)組，元素類型并不一定是 int8_t 類型的。 contents 不占用結(jié)構(gòu)體的大小，它只作為整數(shù)集合數(shù)據(jù)的首指針。整數(shù)集合中的元素按照從小到大的順序在 contents 中排列起來。

2、編碼方式

? ? ? ? 首先，我們來理解編碼方式 encoding 的含義。需要明確的一點是，對于一個整數(shù)集合來說，所有的元素的編碼一定是一致的（否則每個數(shù)都得存一個編碼，而不是將它存在 intset 結(jié)構(gòu)體內(nèi)了），那么整個整數(shù)集合的編碼取決于集合中“絕對值”最大的那個數(shù)（之所以是絕對值，因為整數(shù)包含正數(shù)和負數(shù)）。
? ? ? ??通過那個絕對值最大的整數(shù)來獲取編碼，實現(xiàn)如下：

static uint8_t _intsetValueEncoding(int64_t v) {
    if (v < INT32_MIN || v > INT32_MAX)
        return INTSET_ENC_INT64;
    else if (v < INT16_MIN || v > INT16_MAX)
        return INTSET_ENC_INT32;
    else
        return INTSET_ENC_INT16;
}

? ? ? ? 這段代碼的含義是，如果整數(shù) v 不能用 32位整數(shù)表示，那么就需要用?INTSET_ENC_INT64 編碼；如果不能用 16位整數(shù)表示，那么就需要用?INTSET_ENC_INT32 編碼；否則，采用?INTSET_ENC_INT16 編碼就行。核心就是：能用2個字節(jié)表示就不用4個字節(jié)，能用4個字節(jié)表示就不用8個字節(jié)，能省則省。
? ? ? ? 幾個宏定義在 stdint.h 中，如下：

/* Minimum of signed integral types. */ 
# define INT16_MIN      (-32767-1)  
# define INT32_MIN      (-2147483647-1)  
 
/* Maximum of signed integral types. */  
# define INT16_MAX      (32767)  
# define INT32_MAX      (2147483647)  

3、編碼升級

? ? ? ? 當前編碼方式不足以存儲更大位數(shù)的整數(shù)時，需要升級編碼。舉個例子，下圖所示的四個數(shù)字都在 [ -32768, 32767 ] 范圍內(nèi)，所以采用 INTSET_ENC_INT16 編碼即可。contents 的數(shù)組長度為 sizeof(int16_t) * 4 = 2 * 4 = 8 個字節(jié) （ 即64個二進制位 ）。

? ? ? ? 然后我們插入一個數(shù)，它的值為 32768，比 INT16_MAX 大1，所以它需要采用 INTSET_ENC_INT32 編碼，而整數(shù)集合中所有的數(shù)的編碼需要保持一致。那么，所有數(shù)的編碼都需要轉(zhuǎn)為?INTSET_ENC_INT32 編碼。這就是 “升級”。如圖所示：

? ? ? ? 升級完后，contents 數(shù)組的長度變?yōu)?sizeof(int32_t) * 5?= 4?* 5?= 20?個字節(jié) （ 即160個二進制位 ）。而且每個元素占用的內(nèi)存都擴大一倍，所在的相對位置也發(fā)生了變化，導致所有的元素都需要往高位內(nèi)存遷移。
? ? ? ? 那我們一開始就把所有的整數(shù)集合都用 INTSET_ENC_INT64 來編碼不就好了，還省得麻煩。原因是 Redis 設(shè)計 intset 的初衷還是為了節(jié)省內(nèi)存，設(shè)想一個集合的元素永遠都不會超過 16位 整數(shù)，那么用 64位整數(shù)的話，相當于浪費了 3倍 的內(nèi)存。

三、整數(shù)集合常用操作

1、創(chuàng)建集合

? ? ? ? 創(chuàng)建一個整數(shù)集合?intsetNew，實現(xiàn)在 intset.c 中：

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

? ? ? ?初始創(chuàng)建的整數(shù)集合為空集合，用 zmalloc 進行內(nèi)存分配后，定義編碼為 INTSET_ENC_INT16，這樣可以使內(nèi)存盡量小。這里需要注意的是，intset 的存儲直接涉及到內(nèi)存編碼，所以需要考慮主機的字節(jié)序問題（相關(guān)資料請參閱：字節(jié)序）。
? ? ? ?intrev32ifbe 的意思是 int32 reversal if big endian。即 如果當前主機字節(jié)序為大端序，那么將它的內(nèi)存存儲進行翻轉(zhuǎn)操作。簡言之，intset 的所有成員存儲方式都采用小端序。所以創(chuàng)建一個空的整數(shù)集合，內(nèi)存分布如下：

? ? ? ?了解了整數(shù)集合的內(nèi)存編碼以后，我們來看看它的 設(shè)置 （set）和 獲?。╣et）。

2、元素設(shè)置

? ? ? ?設(shè)置 的含義就是給定整數(shù)集合以及一個位置和值，將值設(shè)置到這個整數(shù)集合的對應位置上。_intsetSet 實現(xiàn)如下：

static void _intsetSet(intset *is, int pos, int64_t value) {
    uint32_t encoding = intrev32ifbe(is->encoding);          /* a */
 
    if (encoding == INTSET_ENC_INT64) {
        ((int64_t*)is->contents)[pos] = value;               /* b */
        memrev64ifbe(((int64_t*)is->contents)+pos);          /* c */
    } else if (encoding == INTSET_ENC_INT32) {
        ((int32_t*)is->contents)[pos] = value;
        memrev32ifbe(((int32_t*)is->contents)+pos);
    } else {
        ((int16_t*)is->contents)[pos] = value;
        memrev16ifbe(((int16_t*)is->contents)+pos);
    }
}

? ? ? ?a) 大端序和小端序只是存儲方式，encoding 在存儲的時候進行了一次 intrev32ifbe 轉(zhuǎn)換，取出來用的時候需要再進行一次 intrev32ifbe 轉(zhuǎn)換（其實就是序列化和反序列化）。
? ? ? ?b) 根據(jù) encoding 的類型，將 contents 轉(zhuǎn)換成指定類型的指針，然后用 pos 進行索引找到對應的內(nèi)存位置，然后將 value 的值設(shè)置到對應的內(nèi)存中。
? ? ? ?c) memrev64ifbe 的實現(xiàn)參見?字節(jié)序 的 memrev64 函數(shù)，即將對應內(nèi)存的值轉(zhuǎn)換成小端序存儲。

3、元素獲取

? ? ? ?獲取 的含義就是給定整數(shù)集合以及一個位置，返回給定位置的元素的值。_intsetGet 實現(xiàn)如下：

static int64_t _intsetGetEncoded(intset *is, int pos, uint8_t enc) {
    int64_t v64;
    int32_t v32;
    int16_t v16;
 
    if (enc == INTSET_ENC_INT64) {
        memcpy(&v64,((int64_t*)is->contents)+pos,sizeof(v64));   /* a */
        memrev64ifbe(&v64);                                      /* b */
        return v64;
    } else if (enc == INTSET_ENC_INT32) {
        memcpy(&v32,((int32_t*)is->contents)+pos,sizeof(v32));
        memrev32ifbe(&v32);
        return v32;
    } else {
        memcpy(&v16,((int16_t*)is->contents)+pos,sizeof(v16));
        memrev16ifbe(&v16);
        return v16;
    }
}
 
static int64_t _intsetGet(intset *is, int pos) {
    return _intsetGetEncoded(is,pos,intrev32ifbe(is->encoding));
}

? ? ? ?a) 根據(jù) encoding 的類型，將 contents 轉(zhuǎn)換成指定類型的指針，然后用 pos 進行索引找到對應的內(nèi)存位置，將內(nèi)存位置上的值拷貝到臨時變量中；
? ? ? ?b) 由于是直接的內(nèi)存拷貝，所以取出來的值還是小端序的，那么在大端序的主機上得到的值是不對的，所以需要再做一次?memrev64ifbe 轉(zhuǎn)換將值還原。

?4、元素查找

? ? ? ?由于整數(shù)集合是有序集合，所以查找某個元素是否在整數(shù)集合中，Redis 采用的是二分查找。intsetSearch 實現(xiàn)如下：

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;                                        /* a */
        return 0;
    } else {                                                      /* b */
        if (value > _intsetGet(is,intrev32ifbe(is->length)-1)) {
            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;       /* c */
        cur = _intsetGet(is,mid);
        if (value > cur) {
            min = mid+1;
        } else if (value < cur) {
            max = mid-1;
        } else {
            break;
        }
    }
    if (value == cur) {                                           /* d */
        if (pos) *pos = mid;
        return 1;
    } else {
        if (pos) *pos = min;
        return 0;
    }
}

? ? ? ?a) 整數(shù)集合為空，返回0表示查找失?。?br>? ? ? ?b) value 的值比整數(shù)集合中的最大值還大，或者比最小值還小，則返回0表示查找失?。?br>? ? ? ?c) 執(zhí)行二分查找，將找到的值存在 cur 中；
? ? ? ?d) 如果找到則返回1，表示查找成功，并且將 pos 設(shè)置為 mid 并返回；如果沒找到則返回一個需要插入的位置。

5、內(nèi)存重分配

? ? ? ?由于 contents 的內(nèi)存是動態(tài)分配的，所以每次進行元素插入或者刪除的時候，都需要重新分配內(nèi)存，這個實現(xiàn)放在 intsetResize 中，實現(xiàn)如下：

static intset *intsetResize(intset *is, uint32_t len) {
    uint32_t size = len*intrev32ifbe(is->encoding);
    is = zrealloc(is,sizeof(intset)+size);
    return is;
}

? ? ? ?encoding 本身表示字節(jié)個數(shù)，所以乘上集合個數(shù) len 就是 contents 數(shù)組需要的總字節(jié)數(shù)了，調(diào)用 zrealloc 進行內(nèi)存重分配，然后返回重新分配后的地址。
? ? ? ?注意：zrealloc 的返回值必須返回出去，因為 intset 在進行內(nèi)存重分配以后，地址可能就變了。即 is = zrealloc(is, ...) 中，此 is 非彼 is。所以，所有調(diào)用 intsetResize 的函數(shù)都需要連帶的返回新的 intset 指針。

6、編碼升級

? ? ? ?編碼升級一定發(fā)生在元素插入，并且插入的元素的絕對值比整數(shù)集合中的元素都大的時候，所以我們把升級后的元素插入和編碼升級放在一個函數(shù)實現(xiàn)，名曰?intsetUpgradeAndAdd，實現(xiàn)如下：

static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
    uint8_t curenc = intrev32ifbe(is->encoding);
    uint8_t newenc = _intsetValueEncoding(value);                             
    int length = intrev32ifbe(is->length);
    int prepend = value < 0 ? 1 : 0;                                         /* a */
    is->encoding = intrev32ifbe(newenc);
    is = intsetResize(is,intrev32ifbe(is->length)+1);                        /* b */
    while(length--)
        _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));   /* c */
    if (prepend)
        _intsetSet(is,0,value);
    else
        _intsetSet(is,intrev32ifbe(is->length),value);                       /* d */
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

? ? ? ?a) curenc 記錄升級前的編碼，newenc 記錄升級后的編碼；
? ? ? ?b) 將整數(shù)集合 is 的編碼設(shè)置成新的編碼后，進行內(nèi)存重分配；
? ? ? ?c) 獲取原先內(nèi)存中的數(shù)據(jù)，設(shè)置到新內(nèi)存中（注意：由于兩段內(nèi)存空間是重疊的，而且新內(nèi)存的長度一定大于原先內(nèi)存，所以需要從后往前進行拷貝）；
? ? ? ?d) 當插入的值 value 為負數(shù)的時候，為了保證集合的有序性，需要插入到 contents 的頭部；反之，插入到尾部；當 value 為負數(shù)時 prepend 為1，這樣就可以保證在內(nèi)存拷貝的時候?qū)⒌?0 個位置留空。
? ? ? ?如圖展示了一個 (-32768, 0, 1, 32767) 的整數(shù)集合在插入數(shù)字 32768 后的升級的完整過程：

? ? ? ? 整數(shù)集合升級的時間復雜度是 O(n) 的，但是在整數(shù)集合的生命期內(nèi)，升級最多發(fā)生兩次（從 INTSET_ENC_INT16 到 INTSET_ENC_INT32 以及 從 INTSET_ENC_INT32 到 INTSET_ENC_INT64）。

7、內(nèi)存遷移

? ? ? ? 絕大多數(shù)情況都是在執(zhí)行 插入 、刪除 、查找 操作。插入 和 刪除 會涉及到連續(xù)內(nèi)存的移動。Redis 的內(nèi)部實現(xiàn)中有一個函數(shù) intsetMoveTail 就是用來實現(xiàn)內(nèi)存移動的。

static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) {
    void *src, *dst;
    uint32_t bytes = intrev32ifbe(is->length)-from;   /* a */
    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);                     /* b */
    } else if (encoding == INTSET_ENC_INT32) {
        src = (int32_t*)is->contents+from;
        dst = (int32_t*)is->contents+to;
        bytes *= sizeof(int32_t);
    } else {
        src = (int16_t*)is->contents+from;
        dst = (int16_t*)is->contents+to;
        bytes *= sizeof(int16_t);
    }
    memmove(dst,src,bytes);                           /* c */
}

? ? ? ?a) 統(tǒng)計從 from 到結(jié)尾，有多少個元素；
? ? ? ?b) 根據(jù)不同的編碼，計算出需要拷貝的內(nèi)存字節(jié)數(shù) bytes，以及拷貝源位置 src，拷貝目標位置 dst；
? ? ? ?c) memmove 是 string.h 中的函數(shù)：src指向的內(nèi)存區(qū)域拷貝 bytes 個字節(jié)到 dst 所指向的內(nèi)存區(qū)域，這個函數(shù)是支持內(nèi)存重疊的；

8、元素插入

? ? ? ?最后，講整數(shù)集合的插入和刪除，插入調(diào)用的是?intsetAdd，在 intset.c 中實現(xiàn)：

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)) {                               /* a */
        return intsetUpgradeAndAdd(is,value);
    } else {
        if (intsetSearch(is,value,&pos)) {                                
            if (success) *success = 0;                                       /* b */
            return is;
        }
        is = intsetResize(is,intrev32ifbe(is->length)+1);                    /* c */
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);    /* d */
    }
    _intsetSet(is,pos,value);                                                 
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);                   /* e */
    return is;
}

? ? ? ?a)?插入的數(shù)值 value 的內(nèi)存編碼大于現(xiàn)有集合的編碼，直接調(diào)用?intsetUpgradeAndAdd 進行編碼升級；
? ? ? ?b) 集合元素是不重復的，如果 intsetSearch 能夠找到，則將 success 置為0，表示此次插入失?。?br>? ? ? ?c) 如果 intsetSearch 找不到，將 intset 進行內(nèi)存重分配，即 長度 加 1。
? ? ? ?d) pos 為?intsetSearch 過程中找到的 value 將要插入的位置，我們將 pos 以后的內(nèi)存向后移動1個單位 （這里的1個單位可能是2個字節(jié)、4個字節(jié)或者8個字節(jié)，取決于當前整數(shù)集合的內(nèi)存編碼）。
? ? ? ?e) 調(diào)用 _intsetSet 將 value 的值設(shè)置到 pos 的位置上，然后給成員變量 length 加?1。最后返回 intset 指針首地址，因為其間進行了 intsetResize，傳入的 intset 指針和返回的有可能不是同一個了。

?9、元素刪除

? ? ? ?刪除元素調(diào)用的是?intsetRemove ，實現(xiàn)如下：

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)) {  /* a */
        uint32_t len = intrev32ifbe(is->length);
        if (success) *success = 1;
        if (pos < (len-1)) intsetMoveTail(is,pos+1,pos);                        /* b */
        is = intsetResize(is,len-1);                                            /* c */
        is->length = intrev32ifbe(len-1); 
    }
    return is;
}

? ? ? ?a) 當整數(shù)集合中存在 value 這個元素時才能執(zhí)行刪除操作；
? ? ? ?b) 如果能通過 intsetSearch 找到元素，那么它的位置就在 pos 上，這是通過 intsetMoveTail 將內(nèi)存往前挪；
? ? ? ?c) intsetResize 重新分配內(nèi)存，并且將集合長度減1；