【Redis5源码学习】2019-04-18 整数集合intset

Grape
全部视频：https://segmentfault.com/a/11...

intset是Redis中的一种数据结构，地位和ziplist,dict一般。

intset的定义？

intset是Redis集合的底层实现之一，当添加的所有数据都是整数时，会使用intset；否则使用dict。特别的，当遇到添加数据为字符串，即不能表示为整数时，Redis 会把数据结构转换为 dict，即把 intset 中的数据全部搬迁到 dict。

intset存在的意义？

intset将整数元素按顺序存储在数组里，并通过二分法降低查找元素的时间复杂度。数据量大时，依赖于“查找”的命令（如SISMEMBER）就会由于O(logn)的时间复杂度而遇到一定的瓶颈，所以数据量大时会用dict来代替intset。但是intset的优势就在于比dict更省内存，而且数据量小的时候O(logn)未必会慢于O(1)的hash function。这也是intset存在的原因。

intset为什么更省内存？

首先，我们看一下Intset的结构体：

typedef struct intset {
uint32_t encoding;   //intset的类型编码
uint32_t length;      //成员元素的个数
int8_t contents[];    //用来存储成员的柔性数组
} intset;

然后对比dict的结构体：

typedef struct dict {
dictEntry **table;
dictType *type;
unsigned long size;
unsigned long sizemask;
unsigned long used;
void *privdata;
} dict;

//需要注意contents数组成员被声明为int8_t类型并不表示contents里存的是int8_t类型的成员，这个类型声明对于contents来说可以    
//认为是毫无意义的，因为intset成员是什么类型完全取决于encoding变量的值。encoding提供下面三种值：
/* Note that these encodings are ordered, so:
 * INTSET_ENC_INT16 < INTSET_ENC_INT32 < INTSET_ENC_INT64. */
#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。

观察intset和dict结构体的空间大小，结果显而易见。在数据量小的时候O(logn)未必会慢于O(1）的hash，所以intset的存在也变得必要。
另外，我们看intset的结构体，观察源码我们知道encoding的种类只有三种，用uint32存储好像有点浪费空间，那么我们在构建结构体的时候是否可以再省一些空间呢？笔者简单抛个砖，把结构体构造成以下这种结构体:

typedef struct intset {
        uint32_t length;
        uint8_t encoding;
        int8_t contents[];
        } intset;

大家觉得是否可以呢？

另外，在此处我们要注意一个问题，intset不管在什么机器上都按照同一种字节序（小端）在内存中存储intset的成员变量。为什么呢？
如果老老实实通过contents[x]的方式赋值取值，我们就不需要考虑这个字节序的问题，但是intset根据encoding的值指定元素的地址偏移，暴力地对内存进行操作。若数据被截断了，则大端机器和小端机器会表现出不统一的状况。为了避免这种情况发生，intset不管在什么机器上都按照同一种字节序(小端)在内存中存intset的成员变量。

那么什么情况下会出现元素的地址偏移呢？不要着急，我们在下文intset的操作的时候会看到，要注意观察哦。

inset的骚操作？

首先我们可以从源码中看到intset的一系列操作：

intset *intsetNew(void);
    intset *intsetAdd(intset *is, int64_t value, uint8_t *success);
    intset *intsetRemove(intset *is, int64_t value, int *success);
    uint8_t intsetFind(intset *is, int64_t value);
    int64_t intsetRandom(intset *is);
    uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value);
    uint32_t intsetLen(const intset *is);
    size_t intsetBlobLen(intset *is);

那么限于篇幅，笔者就拿插入来具体分析，其他若感兴趣可自行查看源码。
对于intset的插入有两种情况，分别为：

1. 插入的value的encoding大于要插入的intset的encoding
2. 插入的value的encoding小于要插入的intset的encoding

   如果是第一种情况，若value的encoding大于要插入的intset的encoding，则调用intsetUpgradeAndAdd直接升级intset的encoding并插入到首部或者尾部。若value的encoding小于要插入的intset的encoding，则不需要升级intset的encoding，调用intsetSearch找到合适的插入位置，再将该位置到contents尾部的数据全部右移一格，最后将value插入到pos。
   是的，很简单，在插入元素的时候比较插入值的encoding和现有的encoding的值，若小于，自己查询位置插入，否则就升级intset插入首部和尾部。对于查询这块，底层用的是二分查找，感兴趣的读者可以去看一看，而为什么是插入首部和尾部，因为在扩展编码之后可能插入的值为负数。

插入的源代码：

/* 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的encoding
        uint32_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;
        else
            return INTSET_ENC_INT16;
    }
    
    /* Upgrades the intset to a larger encoding and inserts the given integer. *///这个函数执行的前提是value参数的大小超过了当前编码//为is->content重新分配内存并修改编码添加value进这个intsetstatic 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底层实现我们可以发现：基于顺序存储的整数集合 执行一些需要用到查询的命令时 其时间复杂度不会是文档里注明O(1)，在操作一个成员插入，查询的平均时间复杂度会是O(logn)。所以当整数集合数据量变大的时候，Redis会用dict作为集合的底层实现，将SADD、SREM、SISMEMBER这些命令的时间复杂度降至O(1)，当然，这会比intset消耗更多内存。所以Redis在实现的时候才会在数据量小的时候采用intset。