[大数据]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流程分析

• 插入新元素,导致rehash产生

• ht[1]扩容到合适的大小,设置rehash进度

• ht[0]数组中元素转移到ht[1]
  

• 交换ht[0]和ht[1]指针指向,然后rehash标记设置为-1表示rehash结束

渐进式rehash

上面列出的rehash看上去很好,但是redis没有这样做,因为如果需要迁移元素很多,由于redis单线程的特性,会导致主线程被阻塞,因此redis采用的是渐进式hash,即慢慢的,慢慢的迁移元素。

小结

ZipList(压缩列表)

ZipList是一种特殊的"双端链表",由一系列特殊编码的连续内存块组成。可以在任意一端进行压入/弹出操作,并且该操作的时间复杂度为0(1)。

压缩列表可以包含任意多个节点,每个节点可以保存一个字节数组或者一个整数值。

压缩列表是列表键和哈希键的底层实现之一。当一个列表键只包含少量列表项,并且每个列表项要么就是小整数值,要么就是长度比较短的字符串,那么Redis底层就会使用ziplist存储存储结构。

当一个哈希键只包含少量列表项,并且每个列表项要么就是小整数值,要么就是长度比较短的字符串,那么Redis底层也会使用ziplist存储存储结构。

zipList构成

zipListEntry构成

ZipList中所有存储长度的数值采用小端字节序,即低位字节在前,高位字节在后。

例如: 数值0x1234,采用小端字节序后实际存储值为: 0x3412

encoding编码

例如: 我们要保存字符串"ab"和"bc"

存储长度的数值采用小端字节序表示

最后一种特殊情况: 1111xxxx,可以在后四位xxxx,表示0001-1101范围的大小,即1-13,但是需要减去一,实际可保存0-12的范围.

如果数值大小在1-12区间内,那么采用最后一种特殊编码方式,不需要content属性

例如: 一个ZipList中包含两个整数值: “2"和"5”

连锁更新问题

• 此时,如果我们将一个长度大于254字节的新节点设置插入进来,称为压缩列表头节点,那么旧头节点的pre_entry_len需要扩展到5字节表示新节点的大小.
• 旧节点加上4字节后变成了254,那么后面的节点需要再次扩展…直到某个节点pre_entry_len扩展到5字节后长度并没有超过254为止
  

ZipList这种特殊情况下产生的多次空间扩展操作称之为连续更新。

新增,删除都可能导致连锁更新的发生。

连锁更新虽然复杂度高,会大大降低性能,但是由于产生概率较低,并且及时出现了,只要被更新节点数量不多,性能上不会有太大影响。

小结

• 压缩列表可以看做一种连续内存空间的双向链表
• 列表的节点之间不是通过指针连接,而是记录上一节点和本节点的长度来寻址,内存占用低
• 如果列表数据过多,导致链表过长,可能影响查询性能
• 增或删较大数据时有可能发生连续更新问题

QuickList(快速链表)

why need ?

限制zipList大小

压缩节点

结构

typedef struct quicklist {
    //头节点指针
    quicklistNode *head;
    //尾结点指针
    quicklistNode *tail;
    //所有zipList的entry的数量
    unsigned long count;        /* total count of all entries in all ziplists */
    //zipLists总数量
    unsigned long len;          /* number of quicklistNodes */
    //zipList的entry上限,默认值 -2 --8kb
    int fill : QL_FILL_BITS;              /* fill factor for individual nodes */
    //首尾不压缩的节点数量
    unsigned int compress : QL_COMP_BITS; /* depth of end nodes not to compress;0=off */
    //内存重分配时的书签数量及数组,一般用不到
    unsigned int bookmark_count: QL_BM_BITS;
    quicklistBookmark bookmarks[];
} quicklist;

typedef struct quicklistNode {
    //前一个节点指针
    struct quicklistNode *prev;
    //下一个节点指针
    struct quicklistNode *next;
    //当前节点的ZipLisr指针
    unsigned char *zl;
    //当前节点的ZipList的字节大小
    unsigned int sz;             /* ziplist size in bytes */
    //当前节点的ZipList的entry个数
    unsigned int count : 16;     /* count of items in ziplist */
    //编码方式: 1.ziplist 2.lzf压缩模式
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    //数据容器类型(预留): 1. 其他 2. zipList
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    //是否被解压缩. 1.则说明被解压了,将来要重新压缩
    unsigned int recompress : 1; /* was this node previous compressed? */
    //测试用
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    //预留字段
    unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode;

• fill为-2表示每个每个ziplist最大内存不超过8kb
• compress为1表示首尾不压缩,中间节点压缩

特点

• 是一个节点为ZipList的双端链表
• 节点采用ZipList,解决了传统链表的内存占用问题
• 控制了ZipList大小,解决了连续内存空间申请效率问题
• 中间节点可以压缩,进一步节省内存

SkipList(跳跃表)

SkipList首先是链表,但与传统链表相比有几点差异:

• 元素按照升序排列存储
• 节点可能包含多个指针,指针跨度不同
  

Redis使用跳跃表作为有序集合键,如果一个有序集合包含的元素数量很多,或者有序集合中元素成员是比较长的字符串，Redis就会使用跳跃表作为有序集合键的底层实现。

例如： sortedSet

Redis目前只在两处地方使用到了SkipList,分别是 :

• 实现有序集合键
• 在集群节点中用作内部数据结构

结构

//t_zset.c
typedef struct zskiplist {
    //头尾节点指针
    struct zskiplistNode *header, *tail;
    //节点数量
    unsigned long length;
    //最大的索引层级,默认为1
    int level;
} zskiplist;

//t_zset.c
typedef struct zskiplistNode {
    //节点存储的值--是sds类型
    sds ele;
    //节点分数--排序,查找用
    double score;
    //前一个节点指针--回退指针
    struct zskiplistNode *backward;
    struct zskiplistLevel {
        //下一个节点指针
        struct zskiplistNode *forward;
        //索引跨度
        unsigned long span;
        //多级索引数组
    } level[];
} zskiplistNode;

特点

• 跳跃表是一个双向链表,每个节点都包含score和ele值
• 节点按照score排序,score值一样则按照ele字典排序
• 每个节点都可以包含多层指针,层数是1到32之间的随机数
• 不同层指针到下一个节点的跨度不同,层级越高,跨度越大
• 增删改成效率与红黑树基本一致,实现却更为简单

RedisObject

Redis中的任意数据类型的键和值都会被封装为一个RedisObject,也叫做Redis对象,源码如下:

• Redis通过引用计数实现了相关内存回收机制,并且还利用该引用计数实现了对象共享机制。
• 通过记录对象最后一次访问时间,可以在服务器启用了maxmemory功能的情况下,将那么较长时间无人访问的键优先淘汰

对象类型与编码

Redis使用对象来表示数据库中的键和值,每次当我们在Redis的数据库中新创建一个键值对时,我们至少会创建两个对象,一个用于做键值对的键,另一个对象做键值对的值。

Reids中会根据存储的数据类型不同，选择不同的编码方式，功包含11种不同的类型:

每种数据类型使用的编码方式如下:

我们可以使用TYPE命令来查看redis中某个键对应的值对象的类型,而不是键对象的类型。

String对象

String是Redis中最常见的数据存储类型:

• 其基本编码方式是RAW,基于简单动态字符串SDS实现,存储上限为512mb.

• 如果存储的SDS长度小于44字节,则会采用EMBSTR编码,此时Object head与SDS是一段连续空间。申请内存时只需要调用一次内存分配函数,效率更高。

内存释放也只需要一次调用

• 如果存储的字符串是整数值,并且大小在LONG—MAX范围内,则会采用INT编码:直接将数据保存在RedisObject的ptr指针位置(刚好8字节),不再需要SDS了

编码的转换

• 如果对保存整数值的字符串对象追加了一个字符串值,那么该字符串对象底层会从int编码转换为raw编码
• 如果对embstr编码的字符串进行修改,那么底层编码也会从embstr转换为raw

List对象

列表对象的编码可以是以下三种:

插入源码分析

/* Implements LPUSH/RPUSH/LPUSHX/RPUSHX. 
 * 'xx': push if key exists. */
//通用的列表插入命令处理
void pushGenericCommand(
    //封装客户端发送来的命令
    client *c, 
    //插入列表头部还是列表尾部
    int where, 
    //是否在Key存在的时候才进行插入操作,默认为false
    //即redis会帮我们自动创建不存在的key
    int xx) {
    int j;
   //redis命令为 lpush key dhy 123 456
   //这里argv[1]拿到的是key
   //redis默认有1-15个db数据库,c->db是去指定的数据库寻找这个key
   //拿到这个key对应的redisObject对象 
    robj *lobj = lookupKeyWrite(c->db, c->argv[1]);
    //该redisObject对象类型必须是OBJ_LIST才可以,如果不是直接返回
    if (checkType(c,lobj,OBJ_LIST)) return;
    //如果拿到的key为null
    if (!lobj) {
        //如果xx为true,说明当可以不存在的时候就不进行处理
        if (xx) {
            addReply(c, shared.czero);
            return;
        }
       //xx默认为false--redis会帮助我们创建一个quicklist对象
        lobj = createQuicklistObject();
        //设置quicklist对象中ziplist的属性
        //限制quicklist中每一个ziplist最大的大小,默认为-2,即8kb
        //是否压缩ziplist,默认为0,不开启压缩
        quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,
                            server.list_compress_depth);
        //执行向数据库db插入key的过程
        dbAdd(c->db,c->argv[1],lobj);
    }
    
    //lpush key dhy 123 456
    //从2开始是value集合
    //把value集合中的元素插入搭配list中
    for (j = 2; j < c->argc; j++) {
        listTypePush(lobj,c->argv[j],where);
        server.dirty++;
    }
    
    addReplyLongLong(c, listTypeLength(lobj));
    
    //发布事件
    char *event = (where == LIST_HEAD) ? "lpush" : "rpush";
    signalModifiedKey(c,c->db,c->argv[1]);
    notifyKeyspaceEvent(NOTIFY_LIST,event,c->argv[1],c->db->id);
}

• 创建quicklist

robj *createQuicklistObject(void) {
    //创建一个quickList
    quicklist *l = quicklistCreate();
    //根据上面的quicklist创建一个redisObject
    robj *o = createObject(OBJ_LIST,l);
    //设置redisObj对象编码为quickList
    o->encoding = OBJ_ENCODING_QUICKLIST;
    return o;
}

Set对象

Set是Redis中的单列集合:

• 不保住有序性
• 包装元素唯一(可以判断元素是否存在)
• 求交集,并集,差集

那什么样的数据类型适合实现set数据结构呢?

HashTable，也就是Redis中的DICT，不过DICT

• 当set第一次被创建时

robj *setTypeCreate(sds value) {
    //判断value是否是数值类型 long long
    if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)
    //如果是整数类型,则采用Intset编码
        return createIntsetObject();
     //否在采用默认编码,也就是HT   
    return createSetObject();
}

• 采用IntSet类型编码

robj *createIntsetObject(void) {
    //初始化INTSET并申请内存空间
    intset *is = intsetNew();
    //创建RedisObject
    robj *o = createObject(OBJ_SET,is);
    //指定编码为INTSET
    o->encoding = OBJ_ENCODING_INTSET;
    return o;
}

• 采用HT编码

robj *createSetObject(void) {
    //初始化DICT类型
    dict *d = dictCreate(&setDictType,NULL);
    //创建redisObj
    robj *o = createObject(OBJ_SET,d);
    //设置编码为HT
    o->encoding = OBJ_ENCODING_HT;
    return o;
}

• 当往set里面添加元素时

/* Add the specified value into a set.
 *
 * If the value was already member of the set, nothing is done and 0 is
 * returned, otherwise the new element is added and 1 is returned. */
int setTypeAdd(robj *subject, sds value) {
    long long llval;
    //如果已经是HT编码,则直接插入元素
    if (subject->encoding == OBJ_ENCODING_HT) {
        dict *ht = subject->ptr;
        dictEntry *de = dictAddRaw(ht,value,NULL);
        if (de) {
            dictSetKey(ht,de,sdsdup(value));
            dictSetVal(ht,de,NULL);
            return 1;
        }
    } 
    //编码为INTSET
    else if (subject->encoding == OBJ_ENCODING_INTSET) {
        //判断编码是否为整数
        if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
        //是整数直接添加到intset里面
            uint8_t success = 0;
            subject->ptr = intsetAdd(subject->ptr,llval,&success);
            if (success) {
            //当intset元素数量超过set_max_intset_entries,则转为HT
                /* Convert to regular set when the intset contains
                 * too many entries. */
                if (intsetLen(subject->ptr) > server.set_max_intset_entries)
                    setTypeConvert(subject,OBJ_ENCODING_HT);
                return 1;
            }
        } else {
        //不是整数,转换为HT
            /* Failed to get integer from object, convert to regular set. */
            setTypeConvert(subject,OBJ_ENCODING_HT);

            /* The set *was* an intset and this value is not integer
             * encodable, so dictAdd should always work. */
            serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK);
            return 1;
        }
    } else {
        serverPanic("Unknown set encoding");
    }
    return 0;
}

set_max_intset_entries默认为512,通过下面的命令可以进行查询

config get set_max_intset_entries

编码转换后:

ZSet对象

因此zset底层将这两个数据结构结合在了一起，具体结构如下:

//zset结构
typedef struct zset {
    //Dict指针  
    dict *dict;
    //SkipList指针
    zskiplist *zsl;
} zset;

创建ZSet对象的方法源码如下:

robj *createZsetObject(void) {
    zset *zs = zmalloc(sizeof(*zs));
    robj *o;
    //创建Dict
    zs->dict = dictCreate(&zsetDictType,NULL);
    //创建SkipList
    zs->zsl = zslCreate();
    //创建zset
    o = createObject(OBJ_ZSET,zs);
    //更改编码
    o->encoding = OBJ_ENCODING_SKIPLIST;
    return o;
}

具体结构如图:

可以当前ZSet最大的问题在于内存的占用过大,因此为了解决这个问题,ZSet提供了两种编码方式,上面给出的是其中一种，适合在是数据量大的情况下使用，发挥出其快速查找的优势

当数据量比较小的时候,ZSet采用ziplist作为底层结构

add源码

void zaddGenericCommand(client *c, int flags) {
    ....
    /* Lookup the key and create the sorted set if does not exist. */
    //zadd添加元素时,先根据key找到zset,不存在则创建新的zset
    zobj = lookupKeyWrite(c->db,key);
    if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
    //判断键是否存在
    if (zobj == NULL) {//不存在
        if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
        if (server.zset_max_ziplist_entries == 0 ||
            server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
        {
        //如果zset_max_ziplist_entries设置为了0就是禁用了ziplist编码
        //或者value大小超过了zset_max_ziplist_value ,采用HT+Skiplist
            zobj = createZsetObject();
        } else {
       //否则采用ziplist
            zobj = createZsetZiplistObject();
        }
        dbAdd(c->db,key,zobj);
    }
...
        int retval = zsetAdd(zobj, score, ele, flags, &retflags, &newscore);
...
}

• createZsetObject

robj *createZsetObject(void) {
//申请内存
    zset *zs = zmalloc(sizeof(*zs));
    robj *o;
//创建Dict
    zs->dict = dictCreate(&zsetDictType,NULL);
//创建SkipList
    zs->zsl = zslCreate();
    o = createObject(OBJ_ZSET,zs);
    o->encoding = OBJ_ENCODING_SKIPLIST;
    return o;
}

• createZsetZiplistObject

robj *createZsetZiplistObject(void) {
//创建ziplist
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_ZSET,zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

• zsetAdd

int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore) {
...
    /* Update the sorted set according to its encoding. */
    //判断编码方式
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {//是ziplist编码
        unsigned char *eptr;
         //判断当前元素是否已经存在,已经存在了则更新score即可   
        if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
             ....
            return 1;
        } else if (!xx) {
            /* Optimize: check if the element is too large or the list
             * becomes too long *before* executing zzlInsert. */ 
            zobj->ptr = zzlInsert(zobj->ptr,ele,score);
            //元素不存在,需要新增,则判断ziplist长度有没有超过限制大小
            //并且元素的大小有无超过限制
            if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries ||
                sdslen(ele) > server.zset_max_ziplist_value)
                //如果超过转换为HT+skipList编码
                zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
            if (newscore) *newscore = score;
            *out_flags |= ZADD_OUT_ADDED;
            return 1;
        } else {
            *out_flags |= ZADD_OUT_NOP;
            return 1;
        }
        //本身就是SkipList+HT编码，无需转换
    } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
            ....
    } else {
        serverPanic("Unknown sorted set encoding");
    }
    return 0; /* Never reached. */
}

ziplist如何做到有序存储

小结

ZSet为了兼顾内存占用和性能,使用了两种编码方式,当数据量小的时候，采用ziplist实现，此时内存占用很小,但是由于数据量也很小，因此性能影响不大。

当数据量增大时,为了性能考虑，采用了HT+SkipList编码实现，此时内存占用很大，但是性能很高。

Hash对象

当超过限制后,底层编码会变成HT

源码

void hsetCommand(client *c) { //客户端相关命令和信息都被封装在了client中
    int i, created = 0;
    robj *o;
    //假设客户端命令为 hset user1 name Jack age 21        
    if ((c->argc % 2) == 1) {
        addReplyErrorFormat(c,"wrong number of arguments for '%s' command",c->cmd->name);
        return;
    }
    //判断hash的key是否存在,不存在就创建一个新的,默认采用ziplist编码
    if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
    //判断是否需要把ziplist转换Dict
    hashTypeTryConversion(o,c->argv,2,c->argc-1);

    for (i = 2; i < c->argc; i += 2)
    //执行插入操作
        created += !hashTypeSet(o,c->argv[i]->ptr,c->argv[i+1]->ptr,HASH_SET_COPY);

...
}

• hashTypeLookupWriteOrCreate—判断hash的key是否存在,不存在就创建一个新的,默认采用ziplist编码

robj *hashTypeLookupWriteOrCreate(client *c, robj *key) {
    //查看key
    robj *o = lookupKeyWrite(c->db,key);
    if (checkType(c,o,OBJ_HASH)) return NULL;
    //不存在,则创建新的
    if (o == NULL) {
        o = createHashObject();
        dbAdd(c->db,key,o);
    }
    return o;
}

• createHashObject—创建一个默认hash对象

robj *createHashObject(void) {
    //默认采用ziplist编码,申请ziplist内存空间
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_HASH, zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

• hashTypeTryConversion—处理hash的编码转换

void hashTypeTryConversion(robj *o, robj **argv, int start, int end) {
    int i;
    //如果编码已经是HT了，那么直接返回,不需要进行编码转换  
    if (o->encoding != OBJ_ENCODING_ZIPLIST) return;
    
    //依次遍历命令中的field,value参数
    for (i = start; i <= end; i++) {
    //如果filed或者value超过hash_max_ziplist_value,则转换为HT
        if (sdsEncodedObject(argv[i]) &&
            sdslen(argv[i]->ptr) > server.hash_max_ziplist_value)
        {
            hashTypeConvert(o, OBJ_ENCODING_HT);
            break;
        }
    }
}

• hashTypeSet–添加元素到hash

int hashTypeSet(robj *o, sds field, sds value, int flags) {
    int update = 0;
     //判断是否为ziplist编码
    if (o->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *zl, *fptr, *vptr;
       
        zl = o->ptr;
        //查询head指针
        fptr = ziplistIndex(zl, ZIPLIST_HEAD);
        //head不为空,说明ziplist不为空,开始查找key
        if (fptr != NULL) {
            fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
            //判断是否存在,如果已经存在则更新
            if (fptr != NULL) {
                /* Grab pointer to the value (fptr points to the field) */
                vptr = ziplistNext(zl, fptr);
                serverAssert(vptr != NULL);
                update = 1;

                /* Replace value */
                zl = ziplistReplace(zl, vptr, (unsigned char*)value,
                        sdslen(value));
            }
        }
         //不存在,则直接push
        if (!update) {
        //依次Push新的field和value到ziplist尾部
            /* Push new field/value pair onto the tail of the ziplist */
            zl = ziplistPush(zl, (unsigned char*)field, sdslen(field),
                    ZIPLIST_TAIL);
            zl = ziplistPush(zl, (unsigned char*)value, sdslen(value),
                    ZIPLIST_TAIL);
        }
        o->ptr = zl;

        /* Check if the ziplist needs to be converted to a hash table */
        //插入了新元素,检查list长度是否超出,超出转化为HT
        if (hashTypeLength(o) > server.hash_max_ziplist_entries)
            hashTypeConvert(o, OBJ_ENCODING_HT);
    } else if (o->encoding == OBJ_ENCODING_HT) {
    //HT编码,直接插入或覆盖
      ...
    } else {
        serverPanic("Unknown hash encoding");
    }

    /* Free SDS strings we did not referenced elsewhere if the flags
     * want this function to be responsible. */
    if (flags & HASH_SET_TAKE_FIELD && field) sdsfree(field);
    if (flags & HASH_SET_TAKE_VALUE && value) sdsfree(value);
    return update;
}

开发者涨薪指南

48位大咖的思考法则、工作方式、逻辑体系