多图解释Redis的整数集合intset升级过程

原创

陈琛

修改于 2020-07-07 14:35:08

5360

修改于 2020-07-07 14:35:08

文章被收录于专栏：陈琛的Redis文章

redis源码分析系列文章

[Redis源码系列]在Liunx安装和常见API

为什么要从Redis源码分析

String底层实现——动态字符串SDS

Redis的双向链表一文全知道

面试官：说说Redis的Hash底层我：......（来自阅文的面试题）

跳跃表确定不了解下😏

前言

大噶好，今天仍然是元气满满的一天，抛开永远写不完的需求，拒绝要求贼变态的客户，单纯的学习技术，感受技术的魅力。（哈哈哈，皮一下很开森）

前面几周我们一起看了Redis底层数据结构，如动态字符串SDS，双向链表Adlist，字典Dict，跳跃表，如果有对Redis常见的类型或底层数据结构不明白的请看上面传送门。

今天来说下set的底层实现整数集合，如果有对set不明白的，常见的API使用这篇就不讲了，看上面的传送门哈。

整数集合概念

整数集合是Redis设计的一种底层结构，是set的底层实现，当集合中只包含整数值元素，并且这个集合元素数据不多时，会使用这种结构。但是如果不满足刚才的条件，会使用其他结构，这边暂时不讲哈。

下图为整数集合的实际组成，包括三个部分，分别是编码格式encoding，包含元素数量length，保存元素的数组contents。（这边只需要简单看下，下面针对每个模块详细说明哈😝）

整数集合的实现

我们看下intset.h里面关于整数集合的定义，上代码哈：

//整数集合结构体
typedef struct intset {
    uint32_t encoding;  //编码格式，有如下三种格式，初始值默认为INTSET_ENC_INT16
    uint32_t length;    //集合元素数量
    int8_t contents[];  //保存元素的数组，元素类型并不一定是ini8_t类型，柔性数组不占intset结构体大小，并且数组中的元素从小到大排列。
} intset;               

#define INTSET_ENC_INT16 (sizeof(int16_t))   //16位，2个字节，表示范围-32,768~32,767
#define INTSET_ENC_INT32 (sizeof(int32_t))   //32位，4个字节，表示范围-2,147,483,648~2,147,483,647
#define INTSET_ENC_INT64 (sizeof(int64_t))   //64位，8个字节，表示范围-9,223,372,036,854,775,808~9,223,372,036,854,775,807

编码格式encoding

包括INTSET_ENC_INT16，INTSET_ENC_INT32，INTSET_ENC_INT64三种类型，其分别对应着不同的范围，具体看上面代码的注释信息。

因为插入的数据的大小是不一样的，为了尽可能的节约内存（毕竟都是钱，平时要省着点用😭），所以我们需要使用不同的类型来存储数据。

集合元素数量length

记录了保存数据contents的长度，即有多少个元素。

保存元素的数组contents

真正存储数据的地方，数组是按照从小到大有序排序的，并且不包含任何重复项（因为set是不含重复项，所以其底层实现也是不含包含项的）。

整数集合升级过程(重点，手动标星)

上面的图我们重新看下，编码格式encoding为INTSET_ENC_INT16，即每个数据占16位。长度length为4，即数组content里面有四个元素，分别是1，2，3，4。如果我们要添加一个数字位40000，很明显超过编码格式为INTSET_ENC_INT16的范围-32,768~32,767，应该是编码格式为INTSET_ENC_INT32。那么他是如何升级的呢，从INTSET_ENC_INT16升级到INTSET_ENC_INT32的呢？

1.了解旧的存储格式

首先我们看下1，2，3，4这四个元素是如何存储的。首先要知道一共有多少位，计算规则为length*编码格式的位数，即4*16=64。所以每个元素占用了16位。

2.确定新的编码格式

新的元素为40000，已经超过了INTSET_ENC_INT16的范围-32,768~32,767，所以新的编码格式为INTSET_ENC_INT32。

3.根据新的编码格式新增内存

上面已经说明了编码格式为INTSET_ENC_INT32，计算规则为length*编码格式的位数，即5*32=160。所以新增的位数为64-159。

4.根据编码格式设置对应的值

从上面知道按照新的编码格式，每个数据应该占用32位，但是旧的编码格式，每个数据占用16位。所以我们从后面开始，每次获取32位用来存储数据。

这样说太难懂了，看下图☺。

首先，那最后32位，即128-159存储40000。那么第49-127是空着的。

接着，取空着的49-127最后的32位，即96到127这32位，用来存储4。那么之前4存储的位置48-63和49-127剩下的64-95这两部分组成了一个大部分，即48-95，现在空着啦。

在接着在48-95这个大部分，再取后32位，即64-95，用来存储3。那么之前3存储位置32-47和48-95剩下的48-63这两部分组成了一个大部分，即32-63，现在空着啦。

再接着，将32-63这个大部分，再取后32位，即还是32-63，用来存储2。那么之前2存储位置16-31空着啦。

最后，将16-31和原来0-31合起来，存储1。

至此，整个升级过程结束。整体来说，分为3步，确定新的编码格式，新增需要的内存空间，从后往前调整数据。

这边有个小问题，为啥要从后往前调整数据呢？

原因是如果从前往后，数据可能会覆盖。也拿上面个例子来说，数据1在0-15位，数据2在16-31位，如果从前往后，我们知道新的编码格式INTSET_ENC_INT32要求每个元素占用32位，那么数据1应该占用0-31，这个时候数据2就被覆盖了，以后就不知道数据2啦。

但是从后往前，因为后面新增了一些内存，所以不会发生覆盖现象。

升级的优点

节约内存

整数集合既可以让集合保存三种不同类型的值，又可以确保升级操作只在有需要的时候进行，这样就节省了内存。

不支持降级

一旦对数组进行升级，编码就会一直保存升级后的状态。即使后面把40000删掉了，编码格式还是不会将会INTSET_ENC_INT16。

整数集合的源码分析

创建一个空集合 intsetnew

这个方法比较简单，是初始化整数集合的步骤，即下图部分。

主要的步骤是分配内存空间，设置默认编码格式，以及初始化数组长度length。

intset *intsetNew(void) {
    intset *is = zmalloc(sizeof(intset));//分配内存空间 
    is->encoding = intrev32ifbe(INTSET_ENC_INT16);//设置默认编码格式INTSET_ENC_INT16 
    is->length = 0;//初始化length 
    return is;
}

添加元素并升级insetAdd流程图（重点）

添加元素并升级insetAdd源码分析

可以根据上面的流程图，对照着下面的源码分析，这边就不写啦哈。

//添加元素
//输入参数*is为原整数集合
//value为要添加的元素
//*success为是否添加成功的标志量 ，1表示成功，0表示失败 
intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
	//确定要添加的元素的编码格式 
    uint8_t valenc = _intsetValueEncoding(value);
    
    uint32_t pos;
    //如果success没有初始值，则初始化为1 
    if (success) *success = 1;

   //如果新的编码格式大于现在的编码格式，则升级并添加元素 
    if (valenc > intrev32ifbe(is->encoding)) {
        //调用另一个方法 
        return intsetUpgradeAndAdd(is,value);
    } else {
        //如果编码格式不变，则调用查询方法 
        //输入参数is为原整数集合 
        //value为要添加的数据
		//pos为位置 
        if (intsetSearch(is,value,&pos)) {//如果找到了，则直接返回，因为数据是不可重复的。 
            if (success) *success = 0;
            return is;
        }

		//设置length 
        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;
}


//#define INT8_MAX 127
//#define INT16_MAX 32767
//#define INT32_MAX 2147483647
//#define INT64_MAX 9223372036854775807LL 
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;
}


//根据输入参数value的编码格式，对整数集合is的编码格式升级 
static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
    //当前集合的编码格式 
	uint8_t curenc = intrev32ifbe(is->encoding);
	//根据对value解析获取新的编码格式 
    uint8_t newenc = _intsetValueEncoding(value);
    //获取集合元素数量 
    int length = intrev32ifbe(is->length);
    //如果要添加的数据小于0，则prepend为1，否则为0 
    int prepend = value < 0 ? 1 : 0;

   //设置集合为新的编码格式，并根据编码格式重新设置内存 
    is->encoding = intrev32ifbe(newenc);
    is = intsetResize(is,intrev32ifbe(is->length)+1);

	//逐步循环，直到length小于0，挨个重新设置每个值，从后往前 
    while(length--)
        _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));

	//如果value为负数，则放在最前面 
    if (prepend)
        _intsetSet(is,0,value);
    else//如果value为整数，设置最末尾的元素为value 
        _intsetSet(is,intrev32ifbe(is->length),value);
    //重新设置length 
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}


//找到is集合中值为value的下标，返回1，并保存在pos中，没有找到返回0，并将pos设置为value可以插入到数组的位置
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;

    //如果集合为空，那么位置pos为0 
    if (intrev32ifbe(is->length) == 0) { 
        if (pos) *pos = 0;
        return 0;
    } else {
        //因为数据是有序集合，如果要添加的数据大于最后一个数字，那么直接把要添加的值放在最后即可，返回最大值下标 
        if (value > _intsetGet(is,intrev32ifbe(is->length)-1)) {
            if (pos) *pos = intrev32ifbe(is->length);
            return 0;
        } else if (value < _intsetGet(is,0)) { //如果这个数据小于数组下标为0的数据，即为最小值 ，返回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;//设置参数pos，返回1，即找到位置 
        return 1;
    } else {//如果没找到，则min和max相邻，随便设置都行，并返回0 
        if (pos) *pos = min; 
        return 0;
    }
}