Redis基础数据结构之 Sorted Set 有序集合 源码解读

Sorted Set 是什么?

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

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

Sorted Set 数据结构

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

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

• 哈希表的高效单点查询特性（ZSCORE）
• 跳表的高效范围查询（ZRANGEBYSCORE）

typedef struct zset {dict *dict;zskiplist *zsl;
} zset;

Skiplist：用于快速查找、插入和删除操作，提供近似O(log N)的时间复杂度

Dictionary（Hashtables）：用来存储成员与分数的映射关系，确保每个成员的唯一性

跳表（skiplist）

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

跳表节点的结构定义

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;

跳表节点查询

在查询某个节点时，跳表会从头节点的最高层开始，查找下一个节点：
访问下一个节点

• 当前节点的元素权重 < 要查找的权重
• 当前节点的元素权重 = 要查找的权重，且节点数据<要查找的数据
  访问当前节点 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;}...
}

层数设置

几种方法：

• 每层的节点数约是下一层节点数的一半。
  • 好处：查找时类似于二分查找，查找复杂度可以减低到 O(logN)
  • 坏处：每次插入/删除节点，都要调整后续节点层数，带来额外开销

随机生成每个节点的层数。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;elsezsl->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
啥都不说了，都在流程图里。

首先判断编码是 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) pairzobj->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 是否合法，不合法直接 returnif (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) pairzobj->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. */
}

Redis 的有序集合通过跳跃表和字典的结合，既保证了成员的唯一性，又提供了高效的排序和检索能力，使其成为处理需要排序数据的理想选择。