Redis 源码简洁剖析 04 - Sorted Set 有序集合

时间:2023-03-08 16:17:21
Redis 源码简洁剖析 04 - Sorted Set 有序集合

Sorted Set 是什么

有序集合(Sorted Set)是 Redis 中一种重要的数据类型,它本身是集合类型,同时也可以支持集合中的元素带有权重,并按权重排序。

  • ZRANGEBYSCORE:按照元素权重返回一个范围内的元素
  • ZSCORE:返回某个元素的权重值

Sorted Set 命令及实现方法

Redis 源码简洁剖析 04 - Sorted Set 有序集合

Sorted Set 数据结构

  • 结构定义:server.h
  • 实现:t_zset.c

结构定义是 zset,里面包含哈希表 dict跳表 zsl。zset 充分利用了:

  • 哈希表的高效单点查询特性(ZSCORE)
  • 跳表的高效范围查询(ZRANGEBYSCORE)
typedef struct zset {
dict *dict;
zskiplist *zsl;
} zset;

跳表(skiplist)

多层的有序链表。下面展示的是 3 层的跳表,头节点是一个 level 数组,作为 level0~level2 的头指针。

Redis 源码简洁剖析 04 - Sorted Set 有序集合

跳表节点的结构定义

typedef struct zskiplistNode {
// sorted set 中的元素
sds ele;
// 元素权重
double score;
// 后向指针(为了便于从跳表的尾节点倒序查找)
struct zskiplistNode *backward;
// 节点的 level 数组
struct zskiplistLevel {
// 每层上的前向指针
struct zskiplistNode *forward;
// 跨度,记录节点在某一层 *forward 指针和该节点,跨越了 level0 上的几个节点
unsigned long span;
} level[];
} zskiplistNode;

跳表的定义

typedef struct zskiplist {
// 头节点和尾节点
struct zskiplistNode *header, *tail;
unsigned long length;
int level;
} zskiplist;

Redis 源码简洁剖析 04 - Sorted Set 有序集合

跳表节点查询

在查询某个节点时,跳表会从头节点的最高层开始,查找下一个节点:

  • 访问下一个节点
    • 当前节点的元素权重 < 要查找的权重
    • 当前节点的元素权重 = 要查找的权重,且节点数据<要查找的数据
  • 访问当前节点 level 数组的下一层指针
    • 当前节点的元素权重 > 要查找的权重
//获取跳表的表头
x = zsl->header;
//从最大层数开始逐一遍历
for (i = zsl->level-1; i >= 0; i--) {
...
while (x->level[i].forward && (x->level[i].forward->score < score || (x->level[i].forward->score == score
&& sdscmp(x->level[i].forward->ele,ele) < 0))) {
...
x = x->level[i].forward;
}
...
}

层数设置

几种方法:

  1. 每层的节点数约是下一层节点数的一半。
    • 好处:查找时类似于二分查找,查找复杂度可以减低到 O(logN)
    • 坏处:每次插入/删除节点,都要调整后续节点层数,带来额外开销
  2. 随机生成每个节点的层数。Redis 跳表采用了这种方法。

Redis 中,跳表节点层数是由 zslRandomLevel 函数决定。

int zslRandomLevel(void) {
int level = 1;
while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

其中每层增加的概率是 0.25,最大层数是 32。

#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^64 elements */
#define ZSKIPLIST_P 0.25 /* Skiplist P = 1/4 */

跳表插入节点 zslInsert

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i, level; serverAssert(!isnan(score));
x = zsl->header;
// 从最高层的 level 开始找
for (i = zsl->level-1; i >= 0; i--) {
// 每层待插入的位置
rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
// forward.score < 待插入 score || (forward.score < 待插入 score && forward.ele < ele)
while (x->level[i].forward &&
(x->level[i].forward->score < score ||
(x->level[i].forward->score == score &&
sdscmp(x->level[i].forward->ele, ele) < 0))) {
// 在同一层 level 找下一个节点
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
} // 随机层数
level = zslRandomLevel(); // 如果待插入节点的随机层数 > 跳表当前的层数
if (level > zsl->level) {
// 增加对应的层数
for (i = zsl->level; i < level; i++) {
rank[i] = 0;
update[i] = zsl->header;
update[i]->level[i].span = zsl->length;
}
zsl->level = level;
}
// 新建节点
x = zslCreateNode(level, score, ele);
// 设置新建节点的 level 数组
for (i = 0; i < level; i++) {
x->level[i].forward = update[i]->level[i].forward;
update[i]->level[i].forward = x; /* update span covered by update[i] as x is inserted here */
x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
update[i]->level[i].span = (rank[0] - rank[i]) + 1;
} for (i = level; i < zsl->level; i++) {
update[i]->level[i].span++;
} x->backward = (update[0] == zsl->header) ? NULL : update[0];
if (x->level[0].forward)
x->level[0].forward->backward = x;
else
zsl->tail = x;
zsl->length++;
return x;
}

跳表删除节点 zslDelete

int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
int i; x = zsl->header;
// 找到待删除的节点
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward &&
(x->level[i].forward->score < score ||
(x->level[i].forward->score == score &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
x = x->level[i].forward;
}
update[i] = x;
}
x = x->level[0].forward;
// 判断节点的 score 和 ele 是否符合条件
if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
// 删除该节点
zslDeleteNode(zsl, x, update);
if (!node)
// 释放内存
zslFreeNode(x);
else
*node = x;
return 1;
}
return 0; /* not found */
}

Sorted Set 基本操作

首先看下如何创建跳表,代码在 object.c 中,可以看到会调用 dictCreate 函数创建哈希表,之后调用 zslCreate 函数创建跳表。

robj *createZsetObject(void) {
zset *zs = zmalloc(sizeof(*zs));
robj *o; zs->dict = dictCreate(&zsetDictType,NULL);
zs->zsl = zslCreate();
o = createObject(OBJ_ZSET,zs);
o->encoding = OBJ_ENCODING_SKIPLIST;
return o;
}

哈希表和跳表的数据必须保持一致。我们通过 zsetAdd 函数研究一下。

zsetAdd

啥都不说了,都在流程图里。

Redis 源码简洁剖析 04 - Sorted Set 有序集合

首先判断编码是 ziplist,还是 skiplist。

ziplist 编码

里面需要判断是否要转换编码,如果转换编码,则需要调用 zsetConvert 转换成 ziplist 编码,这里就不叙述了。

// ziplist 编码时的处理逻辑
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr; // zset 存在要插入的元素
if ((eptr = zzlFind(zobj->ptr, ele, &curscore)) != NULL) {
// 存储要插入的元素时,在 not exist 时更新
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
} ……
if (newscore) *newscore = score; // 原来的 score 和待插入 score 不同
if (score != curscore) {
// 先删除原来的元素
zobj->ptr = zzlDelete(zobj->ptr, eptr);
// 插入新元素
zobj->ptr = zzlInsert(zobj->ptr, ele, score);
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// zset 中不存在要插入的元素
else if (!xx) { // 检测 ele 是否过大 || ziplist 过大
if (zzlLength(zobj->ptr) + 1 > server.zset_max_ziplist_entries ||
sdslen(ele) > server.zset_max_ziplist_value ||
!ziplistSafeToAdd(zobj->ptr, sdslen(ele))) {
// 转换成 skiplist 编码
zsetConvert(zobj, OBJ_ENCODING_SKIPLIST);
} else {
// 在 ziplist 中插入 (element,score) pair
zobj->ptr = zzlInsert(zobj->ptr, ele, score);
if (newscore) *newscore = score;
*out_flags |= ZADD_OUT_ADDED;
return 1;
}
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
}

skiplist 编码

// skiplist 编码时的处理逻辑
if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de; // 从哈希表中查询新增元素
de = dictFind(zs->dict, ele); // 查询到该元素
if (de != NULL) {
/* NX? Return, same element already exists. */
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
} ……
if (newscore) *newscore = score; // 权重发生变化
if (score != curscore) {
// 更新跳表节点
znode = zslUpdateScore(zs->zsl, curscore, ele, score);
// 让哈希表的元素的值指向跳表节点的权重
dictGetVal(de) = &znode->score; /* Update score ptr. */
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// 如果新元素不存在
else if (!xx) {
ele = sdsdup(ele);
// 在跳表中插入新元素
znode = zslInsert(zs->zsl, score, ele);
// 在哈希表中插入新元素
serverAssert(dictAdd(zs->dict, ele, &znode->score) == DICT_OK);
*out_flags |= ZADD_OUT_ADDED;
if (newscore) *newscore = score;
return 1;
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
}

zsetAdd 整体代码

int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore) {
/* Turn options into simple to check vars. */
int incr = (in_flags & ZADD_IN_INCR) != 0;
int nx = (in_flags & ZADD_IN_NX) != 0;
int xx = (in_flags & ZADD_IN_XX) != 0;
int gt = (in_flags & ZADD_IN_GT) != 0;
int lt = (in_flags & ZADD_IN_LT) != 0;
*out_flags = 0; /* We'll return our response flags. */
double curscore; /* NaN as input is an error regardless of all the other parameters. */
// 判断 score 是否合法,不合法直接 return
if (isnan(score)) {
*out_flags = ZADD_OUT_NAN;
return 0;
} /* Update the sorted set according to its encoding. */ // ziplist 编码时的处理逻辑
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr; // zset 存在要插入的元素
if ((eptr = zzlFind(zobj->ptr, ele, &curscore)) != NULL) {
// 存储要插入的元素时,在 not exist 时更新
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
} /* Prepare the score for the increment if needed. */
if (incr) {
score += curscore;
if (isnan(score)) {
*out_flags |= ZADD_OUT_NAN;
return 0;
}
} /* GT/LT? Only update if score is greater/less than current. */
if ((lt && score >= curscore) || (gt && score <= curscore)) {
*out_flags |= ZADD_OUT_NOP;
return 1;
} if (newscore) *newscore = score; // 原来的 score 和待插入 score 不同
if (score != curscore) {
// 先删除原来的元素
zobj->ptr = zzlDelete(zobj->ptr, eptr);
// 插入新元素
zobj->ptr = zzlInsert(zobj->ptr, ele, score);
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// zset 中不存在要插入的元素
else if (!xx) { // 检测 ele 是否过大 || ziplist 过大
if (zzlLength(zobj->ptr) + 1 > server.zset_max_ziplist_entries ||
sdslen(ele) > server.zset_max_ziplist_value ||
!ziplistSafeToAdd(zobj->ptr, sdslen(ele))) {
// 转换成 skiplist 编码
zsetConvert(zobj, OBJ_ENCODING_SKIPLIST);
} else {
// 在 ziplist 中插入 (element,score) pair
zobj->ptr = zzlInsert(zobj->ptr, ele, score);
if (newscore) *newscore = score;
*out_flags |= ZADD_OUT_ADDED;
return 1;
}
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
} /* Note that the above block handling ziplist would have either returned or
* converted the key to skiplist. */ // skiplist 编码时的处理逻辑
if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de; // 从哈希表中查询新增元素
de = dictFind(zs->dict, ele); // 查询到该元素
if (de != NULL) {
/* NX? Return, same element already exists. */
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
} // 从哈希表中查询元素的权重
curscore = *(double *) dictGetVal(de); // 如果要更新元素权重值
if (incr) {
score += curscore;
if (isnan(score)) {
*out_flags |= ZADD_OUT_NAN;
return 0;
}
} /* GT/LT? Only update if score is greater/less than current. */
if ((lt && score >= curscore) || (gt && score <= curscore)) {
*out_flags |= ZADD_OUT_NOP;
return 1;
} if (newscore) *newscore = score; // 权重发生变化
if (score != curscore) {
// 更新跳表节点
znode = zslUpdateScore(zs->zsl, curscore, ele, score);
// 让哈希表的元素的值指向跳表节点的权重
dictGetVal(de) = &znode->score; /* Update score ptr. */
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// 如果新元素不存在
else if (!xx) {
ele = sdsdup(ele);
// 在跳表中插入新元素
znode = zslInsert(zs->zsl, score, ele);
// 在哈希表中插入新元素
serverAssert(dictAdd(zs->dict, ele, &znode->score) == DICT_OK);
*out_flags |= ZADD_OUT_ADDED;
if (newscore) *newscore = score;
return 1;
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
return 0; /* Never reached. */
}

zsetDel

int zsetDel(robj *zobj, sds ele) {
// ziplist 编码
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr; // 找到对应的节点
if ((eptr = zzlFind(zobj->ptr, ele, NULL)) != NULL) {
// 从 ziplist 中删除
zobj->ptr = zzlDelete(zobj->ptr, eptr);
return 1;
}
}
// skiplist 编码
else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
// 从 skiplist 中删除
if (zsetRemoveFromSkiplist(zs, ele)) {
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
return 1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
return 0; /* No such element found. */
}

zsetRemoveFromSkiplist 函数如下:

static int zsetRemoveFromSkiplist(zset *zs, sds ele) {
dictEntry *de;
double score; de = dictUnlink(zs->dict,ele);
if (de != NULL) {
score = *(double*)dictGetVal(de); // 从哈希表 unlink 该元素
dictFreeUnlinkedEntry(zs->dict,de); // 从跳表中删除该元素,并释放内存空间
int retval = zslDelete(zs->zsl,score,ele,NULL);
serverAssert(retval); return 1;
} return 0;
}

代码中的 zslDelete 函数在跳表中分析过(文章中的跳表章节)。

参考链接

Redis 源码简洁剖析系列

最简洁的 Redis 源码剖析系列文章

Java 编程思想-最全思维导图-GitHub 下载链接,需要的小伙伴可以自取~

原创不易,希望大家转载时请先联系我,并标注原文链接。