Redis原理篇之数据结构

Posted 大忽悠爱忽悠

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Redis原理篇之数据结构相关的知识，希望对你有一定的参考价值。

Redis原理篇之数据结构

Redis原理

Redis源码可以去官网下载,也可以从我下面提供的这个链接进行下载:

redis-6.2.4.tar.gz

数据结构

动态字符串SDS

redis中保存的Key是字符串，value大多也是字符串或字符串集合，因此字符串是Redis中最常使用的一种数据结构。

不过Redis没有直接使用C语言中的字符串，因为C语言字符串存在很多问题:

  • 获取字符串长度需要的复杂度为O(N)
  • 非二进制安全,C语言使用空字符’\\0’作为字符串结尾的标记,如果保存的字符串本身含义该标记,那么会造成读取被截断,获取的数据不完整
  • 不可修改
  • 容易造成缓冲区溢出,例如字符串拼接时,超过原本的空间大小,可能会覆盖掉相邻变量的内存空间

而SDS就是对c字符串的封装,以此来解决上述的问题。

SDS结构

SDS是C语言实现的一个结构体:

一个简单的例子如下:

动态扩容

在c语言中,如果要对字符串操作:

  • 拼接–>先进行内存重分配来扩展底层数组大小,如果忘记了这一步,会导致缓冲区溢出
  • 缩短–>需要通过内存重分配来释放字符串不再使用的那部分空间,如果忘记了会导致内存泄露

因为内存重分配需要执行系统调用,并且系统实现内存重分配算法也非常复杂,所以这通过是一个比较耗时的操作

  • 因此通过内存预分配可以减少内存重分配的次数,进而提高整体执行效率

  • 并且SDS还提供了惰性空间释放的功能,即对字符串缩短操作而言,不会立刻使用内存重分配算法来回收多出来的字节,而是通过一个free属性进行记录,当后面需要进行字符串增长时,就会用到

小结

SDS优点如下:

  • O(1)复杂度获取字符串长度
  • 杜绝缓冲区溢出
  • 减少修改字符串长度时所需的内存重分配次数
  • 二进制安全
  • 兼容部分C字符串函数(因此SDS遵循了以’\\0’结尾的惯例)

整数集合IntSet

IntSet是vlaue集合的底层实现之一,当一个集合只包含整数值元素,并且这个集合元素数量不多的情况下,Redis就会使用IntSet作为该value集合的底层实现。

IntSet是Redis用于保存整数值集合抽象数据结构,它可以保存类型为int16_t,int32_t,int64_t的整数值，并且保证集合中不会出现重复元素。

IntSet结构如下:

typedef struct intset 
    //编码方式,支持存放16位,32位,64位整数
    uint32_t encoding;
    //元素个数
    uint32_t length;
    //整数数组,保存集合数据
    int8_t contents[];
 intset;

contents是整数数组底层实现,用来存储元素,并且各个项在数组中的值按从小到大有序排列,并且数组中不包含重复元素。

其中的encoding包含三种模式,表示存储的整数大小不同:

/* Note that these encodings are ordered, so:
 * INTSET_ENC_INT16 < INTSET_ENC_INT32 < INTSET_ENC_INT64. */
/* 2字节整数,范围类似java的short */
#define INTSET_ENC_INT16 (sizeof(int16_t))
/* 4字节整数,范围类似java的int */
#define INTSET_ENC_INT32 (sizeof(int32_t))
/* 8字节整数,范围类似java的long */
#define INTSET_ENC_INT64 (sizeof(int64_t))

为了方便查找,Redis会将intset中所有的整数按照升序依次保存在contents数组中,结构如图:

现在,数组中每个数字都在int16_t的范围内,因此采用的编码方式是INSET_ENC_INT16,每部分占用的字节大小为:

  • encoding: 4字节
  • length: 4字节
  • contents: 2字节*3=6字节

上图中给出的公式是计算每个数组元素起始地址,从这里也能看出为什么很多语言中,数组元素下标都从0开始

因为,如果从1开始,那么公式就变成了: startPtr+(sizeof(int16)*(index-1))

还要额外计算一次减法操作,这会浪费额外的cpu资源

  • startPtr: 数组首元素起始地址
  • sizeof(int16): 数组中每个元素的大小,数组中每个元素大小一致,便于按照下标寻址
  • sizeof(int16)*(index): index下标元素举例起始地址多远,即index元素的起始地址

IntSet升級

  • 升级编码为INTSET_ENC_INT32,每个整数占4字节,并按照新的编码方式及元素个数扩容数组
  • 倒序依次将数组中的元素拷贝到扩容后的正确位置

正序挨个拷贝,会导致前面的元素扩容后覆盖后面的元素,而倒序可以避免这种情况。

c语言写数组插入元素的算法时,也是将元素挨个后移,然后腾出位置,插入新元素。

  • 将待添加的元素放入数组末尾

  • 最后,将intset的encoding属性改为INTSET_ENC_INT32,将length属性改为4

升级源码分析

  • insetAdd–插入元素
/* Insert an integer in the intset */
intset *intsetAdd(
        //需要插入的intset
        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. */
    //判断编码是不是超过了当前intset的编码
    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. */
        //在当前intset中查找值与value一样的元素的角标--使用二分查找法
        //如果找到了,说明元素已经存在,无需再次插入,那么pos就是该元素的位置
        //否则pos指向比value大的前一个元素
        if (intsetSearch(is,value,&pos)) 
            //如果找到了,则无需插入,直接结束并返回
            if (success) *success = 0;
            return is;
        

        //数组扩容
        is = intsetResize(is,intrev32ifbe(is->length)+1);
        //移动数组中pos之后的元素到pos+1,给新元素腾出空间
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
    

    //插入新元素
    _intsetSet(is,pos,value);
    //重置元素长度
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
  • intsetUpgradeAndAdd–升级数组编码
/* Upgrades the intset to a larger encoding and inserts the given integer. */
/* 插入的元素比当前数组编码要大,因此数组需要进行扩容,但是这个新元素具体是插入头部还是尾部不确定
 *  因为该元素可能是一个负数!!!
 * */
static intset *intsetUpgradeAndAdd(intset *is, int64_t value) 
    //获取当intset编码
    uint8_t curenc = intrev32ifbe(is->encoding);
    //获取新编码
    uint8_t newenc = _intsetValueEncoding(value);
    //获取元素个数
    int length = intrev32ifbe(is->length);
    //判断新元素是大于0还是小于0,小于0插入队列头部,大于0插入队尾
    int prepend = value < 0 ? 1 : 0;

    /* 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. */
    //倒序遍历,逐个搬运元素到新的位置,_intsetGetEncoded按照旧编码方式查找旧元素
    while(length--)
        //_intsetSet按照新编码方式将取出的旧元素插入到数组中
        //length+prepend: 如果新元素为负数,那么prepend为1,即旧元素后移的过程中,还会在数组头部腾出一个新位置
        _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));

    /* Set the value at the beginning or the end. */
    //插入新元素,prepend决定是数组头部还是尾部
    if (prepend)
        _intsetSet(is,0,value);
    else
        _intsetSet(is,intrev32ifbe(is->length),value);
    //修改数组长度
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
  • intsetSearch–二分查找元素
/* 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. */
//返回1表示元素存在,我们不需要进行任何操作
//如果返回0,表示元素还不存在
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) 
    //初始化二分查找需要的min,max,mid
    int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;
    //mid对应的值
    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. */
        //数组不为空,判断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;
        
    

    if (value == cur) 
        if (pos) *pos = mid;
        return 1;
     else 
        if (pos) *pos = min;
        return 0;

整数集合升级策略有两个好处:

  • 提升整数集合的灵活性
  • 尽可能节约内存

降级

整数集合不支持降级操作,一旦对数组进行了升级,编码就会一直保持升级后的状态。

内存都是连续存放的,就算进行了降级,也会产生很多内存碎片,如果还要花时间去整理这些碎片更浪费时间。

当然,有小伙伴会说,可以参考SDS的做法,使用free属性来标记空闲空间大小—>当然应该存在更好的做法,大家可以尝试去思考更好的解法

小结

intset具备以下特点:

  • Redis会确保intset中的元素唯一,有序
  • 具备类型升级机制,可以节约内存空间
  • 底层采用二分查找方式来查询

字典(DICT)

Redis是一个键值型(Key-Value Pair)的数据库,我们可以根据键实现快速的增删改查,而键与值的映射关系正是通过Dict实现的。

Dict由三部分组成,分别是: 哈希表(DictHashTable),哈希节点(DictEntry).字典(Dict)

//哈希节点
typedef struct dictEntry 
    //键
    void *key;
    //值
    union 
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
     v;
    //下一个entry的指针
    struct dictEntry *next;
 dictEntry;

/* This is our hash table structure. Every dictionary has two of this as we
 * implement incremental rehashing, for the old to the new table. */
//哈希表
typedef struct dictht 
    //entry数组,数组中保存的是指向entry的指针
    dictEntry **table;
    //哈希表的大小
    unsigned long size;
    //哈希表大小的掩码,总是等于size-1
    unsigned long sizemask;
    //entry的个数
    unsigned long used;
 dictht;

当出现hash碰撞的时候,会采用链表形式将碰撞的元素连接起来,然后链表的新元素采用头插法

//字典
typedef struct dict 
    //dict类型,内置不同的hash函数
    dictType *type;
    //私有数据,在做特殊运算时使用
    void *privdata;
    //一个Dict包含两个哈希表,其中一个是当前数据,另一个一般为空,rehash时使用
    dictht ht[2];
    //rehash的进度,-1表示未开始
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    //rehash是否暂停,1则暂停,0则继续
    int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
 dict;

扩容

/* Expand the hash table if needed */
//如果需要的话就进行扩容
static int _dictExpandIfNeeded(dict *d)

    /* Incremental rehashing already in progress. Return. */
    //如果正在rehash,则返回ok
    if (dictIsRehashing(d)) return DICT_OK;

    /* If the hash table is empty expand it to the initial size. */
    //如果哈希表为空,则初始哈希表为默认大小4
    if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);

    /* If we reached the 1:1 ratio, and we are allowed to resize the hash
     * table (global setting) or we should avoid it but the ratio between
     * elements/buckets is over the "safe" threshold, we resize doubling
     * the number of buckets. */
    //d->ht[0].used >= d->ht[0].size: 说明哈希节点数量已经大于数组长度了,这个条件要满足
    //下面两个条件满足其中一个:
    //1.dict_can_resize: 当服务器执行BGSAVE或者BGREWRITERAO时,该值为假
    //2.d->ht[0].used/d->ht[0].size计算出来的就是负载因子
    //当负载因子大于5时,不管是否正在执行BGSAVE或者BGREWRITERAO,都会进行扩容
    //如果dict type 有expandAllowed函数,则会调用判断是否能够进行扩容
    if (d->ht[0].used >= d->ht[0].size &&
        (dict_can_resize ||
         d->ht[0].used/d->ht[0].size > dict_force_resize_ratio) &&
        dictTypeExpandAllowed(d))
    
        //扩容带下为used+1,底层会对扩容大小进行判断,实际上找的是第一个大于等于used+1的2^n
        return dictExpand(d, d->ht[0].used + 1);
    
    return DICT_OK;

收缩

Dict除了扩容以外,每次删除元素时,也会对负载因子做检查,当LoadFactory<0.1时,会做哈希表收缩:

  • 删除元素源码
/* Delete an element from a hash.
 * Return 1 on deleted and 0 on not found. */
//从hash中删除一个元素,删除成功返回1,没找到返回0
int hashTypeDelete(robj *o, sds field) 
    int deleted = 0;
    //底层采用压缩链表实现,这个暂时不管  
    if (o->encoding == OBJ_ENCODING_ZIPLIST) 
        unsigned char *zl, *fptr;

        zl = o->ptr;
        fptr = ziplistIndex(zl, ZIPLIST_HEAD);
        if (fptr != NULL) 
            fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
            if (fptr != NULL) 
                zl = ziplistDelete(zl,&fptr); /* Delete the key. */
                zl = ziplistDelete(zl,&fptr); /* Delete the value. */
                o->ptr = zl;
                deleted = 1;
            
        
     
    //底层采用hash实现
    else if (o->encoding == OBJ_ENCODING_HT) 
        //删除成功返回C_OK
        if (dictDelete((dict*)o->ptr, field) == C_OK) 
            deleted = 1;

            /* Always check if the dictionary needs a resize after a delete. */
            //删除成功后,检查是否需要重置DICT大小,如果需要则调用dictResize重置
            if (htNeedsResize(o->ptr)) dictResize(o->ptr);
        

     else 
        serverPanic("Unknown hash encoding");
    
    return deleted;
  • htNeedsResize–判断是否需要重置Dict大小
htNeedsResize(dict *dict) 
    long long size, used;
    //哈希表大小--槽的数量就是数组长度
    size = dictSlots(dict);
    //entry数量
    used = dictSize(dict);
    //当哈希表大小大于4并且负载因子低于0.1,表示需要进行收缩
    return (size > DICT_HT_INITIAL_SIZE &&
            (used*100/size < HASHTABLE_MIN_FILL));
  • dictSize–真正进行收缩的源码
/* Resize the table to the minimal size that contains all the elements,
 * but with the invariant of a USED/BUCKETS ratio near to <= 1 */
int dictResize(dict *d)

    unsigned long minimal;
    //如果正在做bgsave或bgrewriteof或rehash,则返回错误
    if (!dict_can_resize || dictIsRehashing(d)) return DICT_ERR;
    //获取entry个数
    minimal = d->ht[0].used;
    //如果entry小于4,则重置为4
    if (minimal < DICT_HT_INITIAL_SIZE)
        minimal = DICT_HT_INITIAL_SIZE;
    //重置大小为minimal,其实是第一个大于等于minimal的2^n
    return dictExpand(d, minimal);

rehash源码分析

  • _dictExpand函数是真正完成扩容的方法,下面来看看这个方法干了啥
/* Expand or create the hash table,
 * when malloc_failed is non-NULL, it'll avoid panic if malloc fails (in which case it'll be set to 1).
 * Returns DICT_OK if expand was performed, and DICT_ERR if skipped. */
int _dictExpand(dict *d, unsigned long size, int* malloc_failed)

    if (malloc_failed) *malloc_failed = 0;

    /* the size is invalid if it is smaller than the number of
     * elements already inside the hash table */
    //如果当前entry数量超过了要申请的size大小,或者正在rehash,直接报错
    if (dictIsRehashing(d) || d->ht[0].used > size)
        return DICT_ERR;
   //声明新的hash table
    dictht n; /* the new hash table */
    //扩容后的数组实际大小,第一个大于等于size的2^n次方
    unsigned long realsize = _dictNextPower(size);

    /* Rehashing to the same table size is not useful. */
    //计算得到的新数组大小与旧数组大小一致,返回错误信息
    if (realsize == d->ht[0].size) return DICT_ERR;

    /* Allocate the new hash table and initialize all pointers to NULL */
    //设置新的hash table的大小和掩码
    n.size = realsize;
    n.sizemask = realsize-1;
    if (malloc_failed) 
        n.table = ztrycalloc(realsize*sizeof(dictEntry*));
        *malloc_failed = n.table == NULL;
        if (*malloc_failed)
            return DICT_ERR;
     else//为新的hash table分配内存: size*entrySize
        n.table = zcalloc(realsize*sizeof(dictEntry*));
    //新的hash table的used为0
    n.used = 0;

    /* Is this the first initialization? If so it's not really a rehashing
     * we just set the first hash table so that it can accept keys. */
    //如果是第一次来,即进行哈希表的初始化,那么直接将
    //上面新创建的n赋值给ht[0]即可
    if (d->ht[0].table == NULL) 
        d->ht[0] = n;
        return DICT_OK;
    

    /* Prepare a second hash table for incremental rehashing */
    //否则,需要rehash，此处需要把rehashidx设置为0
    //表示当前rehash的进度
    //在每次增删改查时都会触发rehash(渐进式hash下面会讲)
    d->ht[1] = n;
    d->rehashidx = 0;
    return DICT_OK;

rehash流程分析