零基础小白？带你阅读Redis源码，从零开始分析Set整数集合模型

之前就说了要来西索Redis，现在来辣！

本文的部分内容参考自《小林Coding》，部分地方根据源代码进行剖析。

Redis源码地址：https://github.com/redis/redis.git

观其面

无序、唯一的键值结合。

这个无序，不是指定没有大小顺序或者字典序，而是不按照插入顺序

一个集合最多可以存储 2^32-1 个元素。概念和数学中个的集合基本类似，可以交集，并集，差集等等，所以 Set 类型除了支持集合内的增删改查，同时还支持多个集合取交集、并集、差集。

Set 类型和 List 类型的区别如下：

• List 可以存储重复元素，Set 只能存储非重复元素；
• List 是按照元素的先后顺序存储元素的，而 Set 则是无序方式存储元素的。

Set的底层数据结构是由哈希表或者证书集合实现的。

• 如果集合中的元素都是整数且元素个数小于 512 （默认值，set-maxintset-entries配置）个，Redis 会使用整数集合作为 Set 类型的底层数据结构；
• 如果集合中的元素不满足上面条件，则 Redis 使用哈希表作为 Set 类型的底层数据结构。

常用操作指令：

# 往集合key中存入元素，元素存在则忽略，若key不存在则新建
SADD key member [member ...]
# 从集合key中删除元素
SREM key member [member ...] 
# 获取集合key中所有元素
SMEMBERS key
# 获取集合key中的元素个数
SCARD key
# 判断member元素是否存在于集合key中
SISMEMBER key member
# 从集合key中随机选出count个元素，元素不从key中删除
SRANDMEMBER key [count]
# 从集合key中随机选出count个元素，元素从key中删除
SPOP key [count]

常用运算指令

# 交集运算
SINTER key [key ...]
# 将交集结果存入新集合destination中
SINTERSTORE destination key [key ...]
# 并集运算
SUNION key [key ...]
# 将并集结果存入新集合destination中
SUNIONSTORE destination key [key ...]
# 差集运算
SDIFF key [key ...]
# 将差集结果存入新集合destination中
SDIFFSTORE destination key [key ...]

应用场景

• 点赞
• 共同关注
• 推荐好友关注
• 抽奖去重

究其身

整数集合

对于哈希，我们在之前已经讲过了，这里我们主要讲讲讲整数集合 intset

typedef struct intset {
   
   
    uint32_t encoding;    // 编码方式
    uint32_t length;    // 集合长度
    int8_t contents[];    // 存储的内容
} intset;

其实这里contents的表观类型是int8_t，其实实际类型由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))

/* Return the required encoding for the provided value. */
// 这里我解释一下这个函数的功能，对value进行编码，也就是encoding的值
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
        // 注意，int8_t在这里也是以int16_t存储的
        return INTSET_ENC_INT16;
}

可能没有看懂，那么我们把宏定义拉出来看看

# define INT32_MAX           INT32_C(2147483647)
# define INT32_MIN         (-INT32_C(2147483647)-1)
# define UINT32_MAX         UINT32_C(4294967295)

#define INT16_MAX            INT16_C(32767)
#define INT16_MIN          (-INT16_C(32767)-1)
#define UINT16_MAX          UINT16_C(65535)

画个数轴你就懂了。

（感谢 ChatGPT 3.5模型 画的图，哈哈哈）

• 如果 encoding 属性值为 INTSET_ENC_INT16，那么 contents 就是一个 int16_t 类型的数组，数组中每一个元素的类型都是 int16_t；
• 如果 encoding 属性值为 INTSET_ENC_INT32，那么 contents 就是一个 int32_t 类型的数组，数组中每一个元素的类型都是 int32_t；
• 如果 encoding 属性值为 INTSET_ENC_INT64，那么 contents 就是一个 int64_t 类型的数组，数组中每一个元素的类型都是 int64_t；

不同类型的 contents 数组，意味着数组的大小也会不同。

数组升级

这里一些简单理解的我就直接复制《小林Codinig》的过来

整数集合会有一个升级规则，就是当我们将一个新元素加入到整数集合里面，如果新元素的类型（int32_t）比整数集合现有所有元素的类型（int16_t）都要长时，整数集合需要先进行升级，也就是按新元素的类型（int32_t）扩展 contents 数组的空间大小，然后才能将新元素加入到整数集合里，当然升级的过程中，也要维持整数集合的有序性。

整数集合升级的过程不会重新分配一个新类型的数组，而是在原本的数组上扩展空间，然后在将每个元素按间隔类型大小分割，如果 encoding 属性值为 INTSET_ENC_INT16，则每个元素的间隔就是 16 位。

举个例子，假设有一个整数集合里有 3 个类型为 int16_t 的元素。

现在，往这个整数集合中加入一个新元素 65535，这个新元素需要用 int32_t 类型来保存，所以整数集合要进行升级操作，首先需要为 contents 数组扩容，在原本空间的大小之上再扩容多 80 位（4x32-3x16=80），这样就能保存下 4 个类型为 int32_t 的元素。

扩容完 contents 数组空间大小后，需要将之前的三个元素转换为 int32_t 类型，并将转换后的元素放置到正确的位上面，并且需要维持底层数组的有序性不变，整个转换过程如下：

整数集合升级有什么好处呢？

如果要让一个数组同时保存 int16_t、int32_t、int64_t 类型的元素，最简单做法就是直接使用 int64_t 类型的数组。不过这样的话，当如果元素都是 int16_t 类型的，就会造成内存浪费的情况。

整数集合升级就能避免这种情况，如果一直向整数集合添加 int16_t 类型的元素，那么整数集合的底层实现就一直是用 int16_t 类型的数组，只有在我们要将 int32_t 类型或 int64_t 类型的元素添加到集合时，才会对数组进行升级操作。

因此，整数集合升级的好处是节省内存资源。

整数集合支持降级操作吗？

不支持降级操作，一旦对数组进行了升级，就会一直保持升级后的状态。比如前面的升级操作的例子，如果删除了 65535 元素，整数集合的数组还是 int32_t 类型的，并不会因此降级为 int16_t 类型。

源码分析

跟着源码来看一看吧，intsetAdd是添加元素的函数，intsetUpgradeAndAdd是扩容并添加的函数，在这之前我要对intrev32ifbe这个宏定义进行一个解释，其实是为了统一编码，如果是小端机器保持不变，打断机器进行mem

前置

/* variants of the function doing the actual conversion only if the target
 * host is big endian */
#if (BYTE_ORDER == LITTLE_ENDIAN)        // 如果当前机器是小端机器
#define memrev16ifbe(p) ((void)(0))
#define memrev32ifbe(p) ((void)(0))
#define memrev64ifbe(p) ((void)(0))
#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


/* Toggle the 16 bit unsigned integer pointed by *p from little endian to
 * big endian */
void memrev16(void *p) {
   
   
    unsigned char *x = p, t;

    t = x[0];
    x[0] = x[1];
    x[1] = t;
}

/* Toggle the 32 bit unsigned integer pointed by *p from little endian to
 * big endian */
void memrev32(void *p) {
   
   
    unsigned char *x = p, t;

    t = x[0];
    x[0] = x[3];
    x[3] = t;
    t = x[1];
    x[1] = x[2];
    x[2] = t;
}

/* Toggle the 64 bit unsigned integer pointed by *p from little endian to
 * big endian */
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;
}

uint16_t intrev16(uint16_t v) {
   
   
    memrev16(&v);
    return v;
}

uint32_t intrev32(uint32_t v) {
   
   
    memrev32(&v);
    return v;
}

uint64_t intrev64(uint64_t v) {
   
   
    memrev64(&v);
    return v;
}

可以看到最后大端机器只是将值和小段进行了交换，对于大小端，这个我们要从《计算机组成原理》来说了，不过这里不细讲，推荐一篇文章：

大端和小端存储模式详解 - Smah - 博客园 (cnblogs.com)

intsetAdd函数和intsetUpgradeAndAdd函数

/* Insert an integer in the intset */
intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
   
   
    uint8_t valenc = _intsetValueEncoding(value);    // 对于要加入的数进行编码
    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)) {
   
   
        /* 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)) {
   
   
            // 如果说这个数字已经存在过了，就退出，并通过success来返回失败状态
            if (success) *success = 0;
            return is;    // 返回一个is，使用责任链模式

        is = intsetResize(is,intrev32ifbe(is->length)+1);    // 修改内存
        // intsetSearch(is,value,&pos)返回的位置在这里用到，如果位置合法就把后面所有元素的位置往后挪，空一格出来
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
    }
    // 把数据放进去
    _intsetSet(is,pos,value);
    // 修改长度
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}


/* Upgrades the intset to a larger encoding and inserts the given integer. */
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;                    // 获取这个值的符号

    /* First set new encoding and resize */
    is->encoding = intrev32ifbe(newenc);                // 修改encoding
    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. */
    // 从后往前一次升级，保证了值不会被覆盖
    // prepend用于确保再开头和结尾都有空格
    while(length--)
        _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;
}

_intsetSet和_intsetGetEncoded函数

/* Set the value at pos, using the configured encoding. */
static void _intsetSet(intset *is, int pos, int64_t value) {
   
   
    uint32_t encoding = intrev32ifbe(is->encoding);
    // _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));
    if (encoding == INTSET_ENC_INT64) {
   
   
        // 修改值
        ((int64_t*)is->contents)[pos] = value;
        // 统一为小端编码
        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, given an encoding. */
// 类型转换
static int64_t _intsetGetEncoded(intset *is, int pos, uint8_t enc) {
   
   
    int64_t v64;
    int32_t v32;
    int16_t v16;
    //_intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc))
    if (enc == INTSET_ENC_INT64) {
   
   
        // 将值复制给v64后返回
        memcpy(&v64,((int64_t*)is->contents)+pos,sizeof(v64));
        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;
    }
}

intsetSearch函数

这个函数用于寻找某个元素是否存在，存在为真，不存在为假，分别返回1和0

/* Search for the position of "value". Return 1 when the value was found and
 * sets "pos" to the position of the value within the intset. Return 0 when
 * the value is not present in the intset and sets "pos" to the position
 * where "value" can be inserted. */
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,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;
    }
}

intsetMoveTail函数

这个函数用于把某个子数组往后面移动to个距离

static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) {
   
   
    // intsetMoveTail(is,pos,pos+1);
    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);
}