Redis原理篇之数据结构
- Redis原理
- 数据结构
- 动态字符串SDS
- SDS结构
- 动态扩容
- 小结
- 整数集合IntSet
- IntSet升級
- 升级源码分析
- 降级
- 小结
- 字典(DICT)
- 扩容
- 收缩
- rehash源码分析
- rehash流程分析
- 渐进式rehash
- 小结
- ZipList(压缩列表)
- zipList构成
- zipListEntry构成
- encoding编码
- 连锁更新问题
- 小结
- QuickList(快速链表)
- why need ?
- 限制zipList大小
- 压缩节点
- 结构
- 特点
- SkipList(跳跃表)
- 结构
- 特点
- RedisObject
- 对象类型与编码
- String对象
- 编码的转换
- List对象
- 插入源码分析
- Set对象
- ZSet对象
- add源码
- ziplist如何做到有序存储
- 小结
- Hash对象
- 源码
Redis原理
Redis源码可以去官网下载,也可以从我下面提供的这个链接进行下载:
redis-6.2.4.tar.gz
数据结构
动态字符串SDS
redis中保存的Key是字符串，value大多也是字符串或字符串集合，因此字符串是Redis中最常使用的一种数据结构。
不过Redis没有直接使用C语言中的字符串，因为C语言字符串存在很多问题:
- 获取字符串长度需要的复杂度为O(N)
- 非二进制安全,C语言使用空字符’ ’作为字符串结尾的标记,如果保存的字符串本身含义该标记,那么会造成读取被截断,获取的数据不完整
- 不可修改
- 容易造成缓冲区溢出,例如字符串拼接时,超过原本的空间大小,可能会覆盖掉相邻变量的内存空间
而SDS就是对c字符串的封装,以此来解决上述的问题。
SDS结构
SDS是C语言实现的一个结构体:
一个简单的例子如下:
动态扩容
在c语言中,如果要对字符串操作:
- 拼接–>先进行内存重分配来扩展底层数组大小,如果忘记了这一步,会导致缓冲区溢出
- 缩短–>需要通过内存重分配来释放字符串不再使用的那部分空间,如果忘记了会导致内存泄露
因为内存重分配需要执行系统调用,并且系统实现内存重分配算法也非常复杂,所以这通过是一个比较耗时的操作
- 因此通过内存预分配可以减少内存重分配的次数,进而提高整体执行效率
- 并且SDS还提供了惰性空间释放的功能,即对字符串缩短操作而言,不会立刻使用内存重分配算法来回收多出来的字节,而是通过一个free属性进行记录,当后面需要进行字符串增长时,就会用到
小结
SDS优点如下:
- O(1)复杂度获取字符串长度
- 杜绝缓冲区溢出
- 减少修改字符串长度时所需的内存重分配次数
- 二进制安全
- 兼容部分C字符串函数(因此SDS遵循了以’ ’结尾的惯例)
整数集合IntSet
IntSet是vlaue集合的底层实现之一,当一个集合只包含整数值元素,并且这个集合元素数量不多的情况下,Redis就会使用IntSet作为该value集合的底层实现。
IntSet是Redis用于保存整数值集合抽象数据结构,它可以保存类型为int16_t,int32_t,int64_t的整数值，并且保证集合中不会出现重复元素。
IntSet结构如下:
代码语言：javascript复制typedef struct intset {
//编码方式,支持存放16位,32位,64位整数
uint32_t encoding;
//元素个数
uint32_t length;
//整数数组,保存集合数据
int8_t contents[];
} intset;contents是整数数组底层实现,用来存储元素,并且各个项在数组中的值按从小到大有序排列,并且数组中不包含重复元素。
其中的encoding包含三种模式,表示存储的整数大小不同:
代码语言：javascript复制/* 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)
代码语言：javascript复制//哈希节点
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;
代码语言：javascript复制当出现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大小
压缩节点
结构
代码语言：javascript复制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,分别是 :
- 实现有序集合键
- 在集群节点中用作内部数据结构
结构
代码语言：javascript复制//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对象
列表对象的编码可以是以下三种:
插入源码分析
代码语言：javascript复制/* 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,通过下面的命令可以进行查询
代码语言：javascript复制config get set_max_intset_entries
编码转换后:
ZSet对象
因此zset底层将这两个数据结构结合在了一起，具体结构如下:
代码语言：javascript复制//zset结构
typedef struct zset {
//Dict指针
dict *dict;
//SkipList指针
zskiplist *zsl;
} zset;创建ZSet对象的方法源码如下:
代码语言：javascript复制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源码
代码语言：javascript复制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
源码
代码语言：javascript复制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;
}
