---

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(//需要插入的intsetintset *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,midint 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-1unsigned 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,则返回okif (dictIsRehashing(d)) return DICT_OK;/* If the hash table is empty expand it to the initial size. *///如果哈希表为空,则初始哈希表为默认大小4if (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^nreturn 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_OKif (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,则重置为4if (minimal < DICT_HT_INITIAL_SIZE)minimal = DICT_HT_INITIAL_SIZE;//重置大小为minimal,其实是第一个大于等于minimal的2^nreturn 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 tabledictht 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*entrySizen.table = zcalloc(realsize*sizeof(dictEntry*));//新的hash table的used为0n.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 --8kbint 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. zipListunsigned 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;//最大的索引层级,默认为1int 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会帮我们自动创建不存在的keyint 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为nullif (!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) {//创建一个quickListquicklist *l = quicklistCreate();//根据上面的quicklist创建一个redisObjectrobj *o = createObject(OBJ_LIST,l);//设置redisObj对象编码为quickListo->encoding = OBJ_ENCODING_QUICKLIST;return o;
}

---

Set对象

Set是Redis中的单列集合:

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

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

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

• 当set第一次被创建时

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

• 采用IntSet类型编码

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

• 采用HT编码

robj *createSetObject(void) {//初始化DICT类型dict *d = dictCreate(&setDictType,NULL);//创建redisObjrobj *o = createObject(OBJ_SET,d);//设置编码为HTo->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;}} //编码为INTSETelse 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;//创建Dictzs->dict = dictCreate(&zsetDictType,NULL);//创建SkipListzs->zsl = zslCreate();//创建zseto = 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,不存在则创建新的zsetzobj = 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+Skiplistzobj = createZsetObject();} else {//否则采用ziplistzobj = 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;
//创建Dictzs->dict = dictCreate(&zsetDictType,NULL);
//创建SkipListzs->zsl = zslCreate();o = createObject(OBJ_ZSET,zs);o->encoding = OBJ_ENCODING_SKIPLIST;return o;
}

• createZsetZiplistObject

robj *createZsetZiplistObject(void) {
//创建ziplistunsigned 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转换DicthashTypeTryConversion(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) {//查看keyrobj *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,则转换为HTif (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不为空,开始查找keyif (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));}}//不存在,则直接pushif (!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长度是否超出,超出转化为HTif (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;
}

---