intset是Redis集合的底層實現之一，當存儲整數集合并且數據量較小的情況下Redis會使用intset作為set的底層實現。當數據量較大或者集合元素為字符串時則會使用dict實現set。
intset將整數元素按順序存儲在數組里，并通過二分法降低查找元素的時間復雜度。數據量大時，依賴于“查找”的命令（如SISMEMBER）就會由于O(logn)的時間復雜度而遇到一定的瓶頸，所以數據量大時會用dict來代替intset。但是intset的優勢就在于比dict更省內存，而且數據量小的時候O(logn)未必會慢于O(1)的hash function。這也是intset存在的原因。

intset結構體聲明

typedef struct intset {uint32_t encoding; //intset的類型編碼uint32_t length; //成員元素的個數int8_t contents[];//用來存儲成員的柔性數組 }

需要注意contents數組成員被聲明為int8_t類型并不表示contents里存的是int8_t類型的成員，這個類型聲明對于contents來說可以認為是毫無意義的，因為intset成員是什么類型完全取決于encoding變量的值。encoding提供下面三種值：

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

如果intset的encoding為INTSET_ENC_INT16，則contents的每個成員的“邏輯類型”都為int16_t。
雖然每個成員的“實際類型”是int8_t，無法直接通過contents[x]取出索引為x的成員元素，但是intset.c里提供了些函數，可以按照不同的encoding方式設置/取出contents的成員。（用指針設置，memcpy取出）
由于這種方法在內存上暴力地賦值與取值，所以希望元素在不同機器上存儲的字節序一致，但是不同處理器 在內存中存放數據的方式不一定相同，主要分為大端字節序和小端字節序。 大小端字節序自己網上搜索資料進行學習。

如果老老實實通過contents[x]的方式賦值取值，我們就不需要考慮這個字節序的問題，但是intset根據encoding的值指定元素的地址偏移，暴力地對內存進行操作。若數據被截斷了，則大端機器和小端機器會表現出不統一的狀況。為了避免這種情況發生，intset不管在什么機器上都按照同一種字節序(小端)在內存中存intset的成員變量。

redis源碼中使用了比較暴力的方式進行大小端的轉換：以64位的轉換為例.

/* variants of the function doing the actual convertion only if the target* host is big endian */ #if (BYTE_ORDER == LITTLE_ENDIAN) #define memrev16ifbe(p) #define memrev32ifbe(p) #define memrev64ifbe(p) #define intrev16ifbe(v) (v) #define intrev32ifbe(v) (v) #define intrev64ifbe(v) (v) #else #define memrev16ifbe(p) memrev16(p) #define memrev32ifbe(p) memrev32(p) #define memrev64ifbe(p) memrev64(p) #define intrev16ifbe(v) intrev16(v) #define intrev32ifbe(v) intrev32(v) #define intrev64ifbe(v) intrev64(v) #endif//翻轉 void memrev64(void *p) {unsigned char *x = p, t;t = x[0];x[0] = x[7];x[7] = t;t = x[1];x[1] = x[6];x[6] = t;t = x[2];x[2] = x[5];x[5] = t;t = x[3];x[3] = x[4];x[4] = t; } uint64_t intrev64(uint64_t v) {memrev64(&v);return v; }

intset基本操作

底層賦值/取值操作

通過_intsetSet和_intsetGet這兩個工具函數，可以根據intset的encoding 讀/寫contents里索引為pos的值。這是后續intset操作的基礎。

/* Set the value at pos, using the configured encoding. */ //按照intset的encoding設置指定位置pos的值 static void _intsetSet(intset *is, int pos, int64_t value) {uint32_t encoding = intrev32ifbe(is->encoding);if (encoding == INTSET_ENC_INT64) {((int64_t*)is->contents)[pos] = value;//大端機器在設置contents時將數據按字節翻轉，按照小端序存儲memrev64ifbe(((int64_t*)is->contents)+pos);} 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);} } /* Return the value at pos, using the configured encoding. */ //按照intset的encoding取出指定位置pos的值 //不對pos進行越界判斷，可能會導致undefined behavior static int64_t _intsetGet(intset *is, int pos) {return _intsetGetEncoded(is,pos,intrev32ifbe(is->encoding)); } /* Return the value at pos, given an encoding. */ //以enc為編碼取出整數集合is在pos索引上的值 //不對pos進行越界判斷，可能會導致undefined behavior 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) {//將contents在pos位置的值賦給v64//不能直接寫contents[pos]的原因是contents時int8_t類型的，contents[pos]表示是以sizeof(int8_t)為單位移動的指針，而實際的編碼是INTSET_ENC_INT64，先將contents指針的類型變為int64_t*memcpy(&v64,((int64_t*)is->contents)+pos,sizeof(v64));//大端機器在取出contents時將原本按照小端序存儲的數據按字節翻轉，讀出正確的值memrev64ifbe(&v64);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;} }

創建一個空intset

空intset的默認encoding是INTSET_ENC_INT16，contents每個成員的邏輯類型是int16_t（雖然還沒有成員）

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

查詢一個成員

前面說了intset是將元素按大小順序存儲在contents數組里，所以在插入新元素之前，必須通過二分法找到合理的插入位置，這由intsetSearch(intset?is, int64_t value, uint32_t?pos)函數實現。
它的作用是在整數集合里用二分法找到value的位置，并把位置寫給pos參數，函數返回1；若沒找到，則寫給pos的是能被插入的value的位置(intset按順序存儲)，函數返回0。

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;/* The value can never be found when the set is empty */if (intrev32ifbe(is->length) == 0) {if (pos) *pos = 0;return 0;} else {/* Check for the case where we know we cannot find the value,* but do know the insert position. *///判斷是否大于小于邊界值if (value > _intsetGet(is,intrev32ifbe(is->length)-1)) {if (pos) *pos = intrev32ifbe(is->length);//value可以被插入的位置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;} }

使用二分法的查找方法還是比較快速的找到對應的值的，上面寫的方式在數據量比較大時，可能會存在越界的可能，可以改為下面的方式：

while (max >= min) {mid = min + (max - min)/2;........ }

原因是：max + min可能就越界了。

插入一個成員

插入一個值為value的成員時，會做以下判斷邏輯：

• 計算value的encoding
• 若value的encoding大于要插入的intset的encoding，則調用intsetUpgradeAndAdd直接升級intset的encoding并插入到首部或者尾部。
• 若value的encoding小于要插入的intset的encoding，則不需要升級intset的encoding，調用intsetSearch找到合適的插入位置，再將該位置到contents尾部的數據全部右移一格，最后將value插入到pos。

/* Insert an integer in the intset */ //success傳null進來則說明外層調用者不需要知道是否插入成功(value是否已存在)，否則success用于此目的 intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {uint8_t valenc = _intsetValueEncoding(value);//根據value的大小計算value的encodinguint32_t pos;if (success) *success = 1;/* Upgrade encoding if necessary. If we need to upgrade, we know that* this value should be either appended (if > 0) or prepended (if < 0),* because it lies outside the range of existing values. */if (valenc > intrev32ifbe(is->encoding)) {//這種插入需要改變encoding(不需要search，因為encoding改變說明value一定插入在contents首部或者尾部)/* This always succeeds, so we don't need to curry *success. */return intsetUpgradeAndAdd(is,value);} else {/* Abort if the value is already present in the set.* This call will populate "pos" with the right position to insert* the value when it cannot be found. */if (intsetSearch(is,value,&pos)) {if (success) *success = 0;//intset里已存在該值，返回失敗return is;}is = intsetResize(is,intrev32ifbe(is->length)+1);if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);//右移一格}_intsetSet(is,pos,value);//插入值is->length = intrev32ifbe(intrev32ifbe(is->length)+1);return is; }/* Return the required encoding for the provided value. */ //根據v值的大小決定需要的編碼類型 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;elsereturn INTSET_ENC_INT16; }/* Upgrades the intset to a larger encoding and inserts the given integer. */ //這個函數執行的前提是value參數的大小超過了當前編碼 //為is->content重新分配內存并修改編碼添加value進這個intset 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;//因為value一定超過了編碼的限制，所以看value是大于0還是小于0以此決定value放置在content[0]還是content[length]/* First set new encoding and resize */is->encoding = intrev32ifbe(newenc);is = intsetResize(is,intrev32ifbe(is->length)+1);/* Upgrade back-to-front so we don't overwrite values.* Note that the "prepend" variable is used to make sure we have an empty* space at either the beginning or the end of the intset. */while(length--)//以curenc為編碼倒序取出所有值并賦值給新的位置_intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));/* Set the value at the beginning or the end. */if (prepend)_intsetSet(is,0,value);else_intsetSet(is,intrev32ifbe(is->length),value);is->length = intrev32ifbe(intrev32ifbe(is->length)+1);return is; }/* Resize the intset */ //解除is的內存分配并重新分配長度為len的intset的內存 static intset *intsetResize(intset *is, uint32_t len) {uint32_t size = len*intrev32ifbe(is->encoding);is = zrealloc(is,sizeof(intset)+size);return is; }//把from索引到intset尾部的整塊數據復制to索引(復制之后from值不變，但是可以被覆蓋) 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) {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); }

移除一個成員

不同于插入一個成員，移除一個成員時不會改變intset的encoding，盡管移除這個成員之后所有成員的encoding都小于所在intset的encoding。

/* Delete integer from intset */ intset *intsetRemove(intset *is, int64_t value, int *success) {uint8_t valenc = _intsetValueEncoding(value);uint32_t pos;if (success) *success = 0;//valenc不可能大于當前編碼，否則value一定不在該intset中if (valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,&pos)) {uint32_t len = intrev32ifbe(is->length);/* We know we can delete */if (success) *success = 1;/* Overwrite value with tail and update length */if (pos < (len-1)) intsetMoveTail(is,pos+1,pos);is = intsetResize(is,len-1);//減小內存分配is->length = intrev32ifbe(len-1);//size-1}return is; }

總結

通過intset底層實現我們可以發現：基于順序存儲的整數集合 執行一些需要用到查詢的命令時 其時間復雜度不會是文檔里注明O(1)，例如：SADD、SREM 操作一個成員時，時間復雜度會是O(logn)。所以當整數集合數據量變大的時候，redis會用dict作為集合的底層實現，將SADD、SREM、SISMEMBER這些命令的時間復雜度降至O(1)，當然，這會比intset消耗更多內存。

總結

以上是生活随笔為你收集整理的Redis之整数集合intset的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。