Redis源码分析之intset篇

intset 是用于实现集合 (set) 这种对外的数据结构。它包含的元素无序，且不能重复。当插入的元素都是整形，底层使用 intset 存储，否则使用 dict。

intset 结构体定义如下：

// intset 结构体
typedef struct intset {
    uint32_t encoding;      // 数据编码，表示 intset 中的每个数据元素用几个字节（2、4、8）来存储，超出上限就需要转 dict 存储
    uint32_t length;        // 表示 intset 中的元素个数
    int8_t contents[];      // 柔性数组 总长度为  encoding * length
} intset;

各字段含义如下：

• encoding 数据编码，表示intset中的每个数据元素用几个字节来存储。

可选值：

#define INTSET_ENC_INT16 (sizeof(int16_t))		// 2字节
#define INTSET_ENC_INT32 (sizeof(int32_t))		// 4字节
#define INTSET_ENC_INT64 (sizeof(int64_t))		// 8字节

• length intset 中的元素个数；
• contents 柔性数组，存放真实数据，总大小为 encoding * length。

需要注意的是，intset 会随着数据的添加而改变它的数据编码：

127.0.0.1:6379> sadd myset 1 2 3
(integer) 3
127.0.0.1:6379> memory usage myset
(integer) 61
127.0.0.1:6379> sadd myset 4 5
(integer) 2
127.0.0.1:6379> memory usage myset
(integer) 65
127.0.0.1:6379> sadd myset 36666
(integer) 1
127.0.0.1:6379> memory usage myset
(integer) 79

我们来分析一下：

添加 3 个小元素后，myset 占用 61 字节，再添加两个小元素，myset 占用 65 字节，说明 1 个小元素占用 2字节，与我们的分析一致。同时也可以得出除了元素所占的空间，结构体本身占用 65 - 5*2 = 55 字节。

接着我们添加了 36666 元素，超出了 2 字节的表示范围，因此变为了4字节编码，共有（1、2、3、4、5、36666）6 个元素，占用 24 字节。

结构体本身 55 字节 + 元素 24 字节 = 79 字节

与输出结果一致。

了解了数据结构后，分析两个主要函数（插入和查找）：

intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
    // 获取要插入 value 的编码类型
    uint8_t valenc = _intsetValueEncoding(value);
    uint32_t pos;
    if (success) *success = 1;

    // 如果比当前的编码类型大，需要进行升级
    if (valenc > intrev32ifbe(is->encoding)) {
        return intsetUpgradeAndAdd(is, value);
    } else {
        // 如果当前 value 已经存在
        if (intsetSearch(is, value, &pos)) {
            if (success) *success = 0;
            return is;
        }

        // 内存扩充以容纳新的元素
        is = intsetResize(is, intrev32ifbe(is->length) + 1);
        // 将待插入位置后面的元素统一向后移动 1 个位置
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is, pos, pos + 1);
    }

    // 把 value 插入到指定 pos
    _intsetSet(is, pos, value);

    // length + 1
    is->length = intrev32ifbe(intrev32ifbe(is->length) + 1);
    return is;
}

• 获取要插入数据的编码类型，与当前编码类型比较：
  • 比当前编码类型大，进行升级存储，同时由于是新元素造成的存储升级，所以很轻松知道要插入元素的位置；
  • 比当前编码类型小，传入 value 进行查找：
    • 能找到，直接返回；
    • 找不到，返回要插入的位置，首先进行内存扩充以容纳新元素，再把要插入位置的后面元素集体后移1个位置，最后把当前 value 插入到指定位置，整体长度 + 1。

所以这个插入方法时间复杂度为 O(N)。

// 在指定的 intset 中查找指定的元素 value，如果找到，则返回 1 并且将参数 pos 指向找到的元素位置
// 如果没找到，则返回 0 并且将参数 pos 指向能插入该元素的位置。
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
        // intset 里的数据是按小端（little endian）模式存储的，因此在大端（big endian）机器上运行时，这里的 intrev32ifbe 会做相应的转换
    int min = 0, max = intrev32ifbe(is->length) - 1, mid = -1;
    
    int64_t cur = -1;

    // 如果 intset 为空,直接返回 0 和要插入的位置 0
    if (intrev32ifbe(is->length) == 0) {
        if (pos) *pos = 0;
        return 0;
    } else {
        // 如果要插入的 value > 最大值 或 < 最小值,那么要插入的位置就确定了
        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;
        }
    }

    // 如果找到的值就是要插入的值，返回1，并把 pos 指向该位置
    if (value == cur) {
        if (pos) *pos = mid;
        return 1;
    } else {    // 返回 0，并把 pos 指向该插入的位置
        if (pos) *pos = min;
        return 0;
    }
}

• 查找时首先对特殊情况进行处理（intset 为空、比最小值小，比最大值大）
• 接着进行二分查找
  • 如果最后找到的值和要插入的值相同，修改 pos 为当前位置，返回 1；
  • 如果不同，说明需要插入，pos 指向要插入的位置，返回 0。

查找时间复杂度为 O(log N)。

Intset 和 ziplist 相比，都有时间换空间的意思，但也有一些不同：

• intset 中每个元素都是定长的，ziplist 元素长度可以不同；
• intset 只能存储整数，而 ziplist 任意二进制串都可以存储；
• intset 中的元素是有序的，查找时可以进行二分查找，而 ziplist 只能遍历。

另外，intset 在以下情况发生时会转变为 dict

• 添加了字符串，intset 无法表示

127.0.0.1:6379> sadd myset 1 2 3
(integer) 3
127.0.0.1:6379> object encoding myset
"intset"
127.0.0.1:6379> sadd myset zwj
(integer) 1
127.0.0.1:6379> object encoding myset
"hashtable

• 添加了一个数字，但它超出了能表示的范围。intset 能够表达的最大的整数范围为 -2^64^~2^64^-1，因此，如果添加的数字超出了这个范围，这也会导致 intset 转成 dict
• 添加的集合元素个数超过了set-max-intset-entries配置的值（默认 512）的时候，此时查找效率为 O(log N)，数据量大时会比较慢，也会导致 intset 转成 dict

Redis 的 set 数据结构还提供了计算交集（sinter）、并集（sunion）、差集（sdiff）的方法：

计算交集的过程大概可以分为三部分：

1. 检查各个集合，对于不存在的集合当做空集来处理。一旦出现空集，则不用继续计算了，最终的交集就是空集；
2. 对各个集合按照元素个数由少到多进行排序。这个排序有利于后面计算的时候从最小的集合开始，需要处理的元素个数较少；
3. 对排序后第一个集合（也就是最小集合）进行遍历，对于它的每一个元素，依次在后面的所有集合中进行查找。只有在所有集合中都能找到的元素，才加入到最后的结果集合中。

O(N*M) worst case where N is the cardinality of the smallest set and M is the number of sets.

计算并集只需要遍历所有集合，将每一个元素都添加到最后的结果集合中。向集合中添加元素会自动去重。

O(N) where N is the total number of elements in all given sets.

计算差集（A 和 B 的差集是 A 中有 B 中没有的，A - B）

计算差集有两种可能的算法，它们的时间复杂度有所区别。

第一种算法：

• 对第一个集合进行遍历，对于它的每一个元素，依次在后面的所有集合中进行查找。只有在所有集合中都找不到的元素，才加入到最后的结果集合中。

这种算法的时间复杂度为 O(N*M)，其中N是第一个集合的元素个数，M是集合数目。

第二种算法：

• 将第一个集合的所有元素都加入到一个中间集合中；
• 遍历后面所有的集合，对于碰到的每一个元素，从中间集合中删掉它；
• 最后中间集合剩下的元素就构成了差集。

这种算法的时间复杂度为O(N)，其中N是所有集合的元素个数总和。

在计算差集的开始部分，会先分别估算一下两种算法预期的时间复杂度，然后选择复杂度低的算法来进行运算。还有两点需要注意：

• 在一定程度上优先选择第一种算法，因为它涉及到的操作比较少，只用添加，而第二种算法要先添加再删除；
• 如果选择了第一种算法，那么在执行该算法之前，Redis 的实现中对于第二个集合之后的所有集合，按照元素个数由多到少进行了排序。这个排序有利于以更大的概率查找到元素，从而更快地结束查找。