Redis有序集合

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

为什么 Sorted Set 能同时提供以下两种操作接口,以及它们的复杂度分别是 O(logN)+M 和 O(1) 呢?

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

实际上,这个问题背后的本质是:为什么 Sorted Set 既能支持高效的范围查询,同时还能以 O(1) 复杂度获取元素权重值?

127.0.0.1:6379> zadd zset_user_1 1 mysql
(integer) 1
127.0.0.1:6379> zadd zset_user_1 2 redis
(integer) 1
127.0.0.1:6379> zadd zset_user_1 3 es
(integer) 1
127.0.0.1:6379> zadd zset_user_1 4 hbase
(integer) 1
127.0.0.1:6379> zadd zset_user_1 5 clickhouse
(integer) 1
127.0.0.1:6379>
127.0.0.1:6379> zrange zset_user_1 0 10
1) "mysql"
2) "redis"
3) "es"
4) "hbase"
5) "clickhouse"
127.0.0.1:6379>
127.0.0.1:6379> zrange zset_user_1 0 10 withscores
 1) "mysql"
 2) "1"
 3) "redis"
 4) "2"
 5) "es"
 6) "3"
 7) "hbase"
 8) "4"
 9) "clickhouse"
10) "5"
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379> zrangebyscore zset_user_1 2 4
1) "redis"
2) "es"
3) "hbase"
127.0.0.1:6379>
127.0.0.1:6379> zrangebyscore zset_user_1 2 4 withscores
1) "redis"
2) "2"
3) "es"
4) "3"
5) "hbase"
6) "4"
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379> zscore zset_user_1 mysql
"1"
127.0.0.1:6379>
127.0.0.1:6379> zscore zset_user_1 es
"3"
127.0.0.1:6379>
127.0.0.1:6379>

这其实就和 Sorted Set 底层的设计实现有关了。
Sorted Set 能支持范围查询,这是因为它的核心数据结构设计采用了跳表,
而它能以常数复杂度获取元素权重,这是因为它同时采用了哈希表进行索引。

Sorted Set 是如何把这两种数据结构结合在一起的?它们又是如何进行协作的呢?

Sorted Set 采用的双索引的设计思想和实现。

(1) 有序集合(Sorted Set)

/**
 * ZSET 是有序集合,使用两个数据结构来保存相同的元素,以便将 O(log(N)) INSERT 和 REMOVE 操作放入已排序的数据结构中。
 * 
 * 这些元素被添加到将 Redis 对象映射到分数的哈希表中。
 * 同时,将元素添加到将分数映射到 Redis 对象的跳跃列表(因此对象在此“视图”中按分数排序)。
 *
 * 请注意,为了节省内存,表示元素的 SDS 字符串在哈希表和跳表中都是相同的。 
 * 为了更容易地管理共享的 SDS 字符串,我们所做的是仅在 zslFreeNode() 中释放 SDS 字符串。 
 * 字典没有设置无值方法。
 * 所以我们应该总是从字典中删除一个元素,然后再从跳过列表中删除。
 */

(1.1) 有序集合结构

/**
 * 有序集合 zset 结构体
 */
typedef struct zset {
 dict *dict; // 字典
 zskiplist *zsl; // 跳表
} zset;

Sorted Set 同时采用跳表哈希表两个索引结构
利用了跳表高效支持范围查询(如 ZRANGEBYSCORE 操作),以及哈希表高效支持单点查询(如 ZSCORE 操作)的特征
可以在一个数据类型中,同时高效支持范围查询和单点查询,这是单一索引结构比较难达到的效果。

跳表或是哈希表中,各自保存了什么样的数据?
跳表和哈希表保存的数据是如何保持一致的?

(1.2) Redis有序集合支持的命令

支持的命令见 https://github.com/redis/redi...

(1.2.1) 新建有序集合

// file: object.c
/**
 * 创建有序集合对象
 * 底层数据结构 跳表+哈希表
 */
robj *createZsetObject(void) {
 // 为zset结构体分配内存
 zset *zs = zmalloc(sizeof(*zs));
 robj *o;
 // 创建字典
 zs->dict = dictCreate(&zsetDictType,NULL);
 // 创建跳表
 zs->zsl = zslCreate();
 // 创建redisObject对象,类型为有序集合
 o = createObject(OBJ_ZSET,zs);
 // 设置编码方式为 跳表
 o->encoding = OBJ_ENCODING_SKIPLIST;
 return o;
}
// file: object.c
/**
 * 创建有序集合压缩列表对象
 * 底层数据结构 压缩列表
 */
robj *createZsetZiplistObject(void) {
 // 创建压缩列表
 unsigned char *zl = ziplistNew();
 // 创建redisObject对象,类型为有序集合
 robj *o = createObject(OBJ_ZSET,zl);
 // 设置编码方式为 压缩列表 
 o->encoding = OBJ_ENCODING_ZIPLIST;
 return o;
}

(1.2.2) 往有序集合添加元素-zsetAdd

// file: 
/**
 * 添加新元素或更新排序集中现有元素的分数,无论其编码如何。
 *
 * 标记会更改命令行为。 它们通过整数指针传递,因为该函数将清除标志并用其他标志填充它们以指示不同的条件。
 *
 * 输入标记如下:
 * ZADD_INCR: 将当前元素分数增加'score',而不是更新当前元素分数。 如果该元素不存在,我们假设以前分数为0。
 * ZADD_NX: 仅当元素不存在时才执行操作。
 * ZADD_XX: 仅当元素不存在时才执行操作。
 *
 * 当使用 ZADD_INCR 时,如果 'newscore' 不为 NULL,则元素的新分数存储在 '*newscore' 中。
 *
 * 返回的标记如下:
 * ZADD_NAN: 结果分数不是数字。
 * ZADD_ADDED: 元素已添加(调用前不存在)。 
 * ZADD_UPDATED: 元素分数已更新。
 * ZADD_NOP: 由于 NX 或 XX,未执行任何操作。
 *
 * 返回值:
 * 该函数在成功时返回 1,并设置适当的标志 ADDED 或 UPDATED 以指示操作期间发生的事情
 * (请注意,如果我们使用与以前相同的分数重新添加元素,则不能设置任何标志,或者在这种情况下 使用零增量)。
 *
 * 该函数在出错时返回 0,目前仅当增量产生 NAN 条件时,或者当“score”值自开始以来为 NAN 时。 
 * 
 * 作为添加新元素的副作用的命令可能会将排序集内部编码从 ziplist 转换为 hashtable + skiplist。 
 *
 * 'ele'的内存管理:
 * 该函数不获取“ele”SDS 字符串的所有权,但会在需要时复制它。
 * 
 * 
 * zadd zset_user_1 1 mysql 2 redis 
 * @param *zobj 有序集合对象
 * @param score 分数 对应 1 2
 * @param ele 元素 其实就是 member 对应的是 mysql redis 
 * @param *flags 
 * @param *newscore 
 */
int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) {
 /* 将选项变成简单的检查变量。 */
 int incr = (*flags & ZADD_INCR) != 0;
 int nx = (*flags & ZADD_NX) != 0;
 int xx = (*flags & ZADD_XX) != 0;
 *flags = 0; /* We'll return our response flags. */
 double curscore;
 /* NaN as input is an error regardless of all the other parameters. */
 // 参数校验
 if (isnan(score)) {
 *flags = ZADD_NAN;
 return 0;
 }
 /* 根据编码更新有序集合。 */
 if (zobj->encoding == OBJ_ENCODING_ZIPLIST) { // 编码是压缩列表
 unsigned char *eptr;
 if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
 /* NX? Return, same element already exists. */
 if (nx) {
 *flags |= ZADD_NOP;
 return 1;
 }
 /* Prepare the score for the increment if needed. */
 if (incr) {
 score += curscore;
 if (isnan(score)) {
 *flags |= ZADD_NAN;
 return 0;
 }
 if (newscore) *newscore = score;
 }
 /* Remove and re-insert when score changed. */
 if (score != curscore) {
 zobj->ptr = zzlDelete(zobj->ptr,eptr);
 zobj->ptr = zzlInsert(zobj->ptr,ele,score);
 *flags |= ZADD_UPDATED;
 }
 return 1;
 } else if (!xx) {
 /* check if the element is too large or the list
 * becomes too long *before* executing zzlInsert. */
 if (zzlLength(zobj->ptr)+1 > server.zset_max_ziplist_entries ||
 sdslen(ele) > server.zset_max_ziplist_value ||
 !ziplistSafeToAdd(zobj->ptr, sdslen(ele)))
 {
 zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
 } else {
 zobj->ptr = zzlInsert(zobj->ptr,ele,score);
 if (newscore) *newscore = score;
 *flags |= ZADD_ADDED;
 return 1;
 }
 } else {
 *flags |= ZADD_NOP;
 return 1;
 }
 }
 /* 请注意,上面的块处理ziplist将返回 或 转换键为skiplist。 
 * Note that the above block handling ziplist would have either returned or converted the key to skiplist. 
 */
 if (zobj->encoding == OBJ_ENCODING_SKIPLIST) { // 编码是跳表
 // 有序集合
 zset *zs = zobj->ptr;
 // 跳表头节点
 zskiplistNode *znode;
 // 字典
 dictEntry *de;
 // ele是member // zscore key member 的时候获取score 用到了这儿
 // 从字典里根据key查到的节点
 de = dictFind(zs->dict,ele);
 if (de != NULL) { // 节点已存在
 /* NX? Return, same element already exists. */
 if (nx) {
 *flags |= ZADD_NOP;
 return 1;
 }
 // 获取节点的值 也就是 分数
 curscore = *(double*)dictGetVal(de);
 /* 如果需要,为增量准备分数。 */
 if (incr) {
 score += curscore;
 if (isnan(score)) {
 *flags |= ZADD_NAN;
 return 0;
 }
 // 更新分值
 if (newscore) *newscore = score;
 }
 /* 当分数改变时移除并重新插入。 */
 if (score != curscore) {
 // 
 znode = zslUpdateScore(zs->zsl,curscore,ele,score);
 /* 请注意,我们没有从表示已排序集合的哈希表中删除原始元素,因此我们只更新分数。 */
 dictGetVal(de) = &znode->score; // 更新分数指针
 // 更新标记
 *flags |= ZADD_UPDATED;
 }
 return 1;
 } else if (!xx) {
 ele = sdsdup(ele);
 znode = zslInsert(zs->zsl,score,ele);
 serverAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
 *flags |= ZADD_ADDED;
 if (newscore) *newscore = score;
 return 1;
 } else {
 *flags |= ZADD_NOP;
 return 1;
 }
 } else {
 serverPanic("Unknown sorted set encoding");
 }
 return 0; /* Never reached. */
}

(1.2.x) zadd命令源码解析

void zaddCommand(client *c) {
 zaddGenericCommand(c,ZADD_NONE);
}
/**
 * This generic command implements both ZADD and ZINCRBY. 
 * 
 * zadd zset_user_1 1 mysql
 */
void zaddGenericCommand(client *c, int flags) {
 static char *nanerr = "resulting score is not a number (NaN)";
 // 输入的key 
 robj *key = c->argv[1];
 robj *zobj;
 sds ele;
 double score = 0, *scores = NULL;
 int j, elements;
 int scoreidx = 0;
 /* 以下变量用于跟踪命令在执行期间实际执行的操作、回复客户端并触发键空间更改通知。 */
 int added = 0; /* 添加的新元素个数 */
 int updated = 0; /* 更新分数的元素个数 */
 int processed = 0; /* 处理的元素个数, 对于 XX 等选项可能会保持为零。 */
 /* 解析选项。 最后,'scoreidx' 设置为第一个分数元素对的分数的参数位置。 */
 scoreidx = 2;
 // 解析后面的参数
 while(scoreidx < c->argc) {
 char *opt = c->argv[scoreidx]->ptr;
 // 位运算 flag是0 对应二进制是 0000 0000 0000 0000 
 // |= 按位或 
 // 如果包含就 执行 按位或运算 
 if (!strcasecmp(opt,"nx")) flags |= ZADD_NX;
 else if (!strcasecmp(opt,"xx")) flags |= ZADD_XX;
 else if (!strcasecmp(opt,"ch")) flags |= ZADD_CH;
 else if (!strcasecmp(opt,"incr")) flags |= ZADD_INCR;
 else break;
 scoreidx++;
 }
 /* 将选项变成简单的检查变量。 */
 // 位运算 
 int incr = (flags & ZADD_INCR) != 0;
 int nx = (flags & ZADD_NX) != 0;
 int xx = (flags & ZADD_XX) != 0;
 int ch = (flags & ZADD_CH) != 0;
 /* 在选项之后,我们希望有偶数个参数,因为我们希望有任意数量的分数元素对。 */
 elements = c->argc-scoreidx;
 if (elements % 2 || !elements) {
 addReply(c,shared.syntaxerr);
 return;
 }
 elements /= 2; /* Now this holds the number of score-element pairs. */
 /* Check for incompatible options. */
 if (nx && xx) {
 addReplyError(c,
 "XX and NX options at the same time are not compatible");
 return;
 }
 if (incr && elements > 1) {
 addReplyError(c,
 "INCR option supports a single increment-element pair");
 return;
 }
 /* Start parsing all the scores, we need to emit any syntax error
 * before executing additions to the sorted set, as the command should
 * either execute fully or nothing at all. */
 scores = zmalloc(sizeof(double)*elements);
 for (j = 0; j < elements; j++) {
 if (getDoubleFromObjectOrReply(c,c->argv[scoreidx+j*2],&scores[j],NULL)
 != C_OK) goto cleanup;
 }
 /* Lookup the key and create the sorted set if does not exist. */
 // 查找键并创建排序集(如果不存在)
 // 根据key查找value
 zobj = lookupKeyWrite(c->db,key);
 // zobj为空,表示有序集合还未创建
 if (zobj == NULL) {
 if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
 // 服务器配置里 zset_max_ziplist_entries==0 或 zset_max_ziplist_value < 每个元素的长度 
 if (server.zset_max_ziplist_entries == 0 ||
 server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
 {
 // 创建 Zset对象,底层数据结构是跳表和哈希表 
 zobj = createZsetObject();
 } else {
 // 也就是 同时满足以下3个条件,会使用 ziplist 作为 有序集合的底层编码方式 
 // 压缩列表为空时,
 // 并且 服务器配置里 有序集合最大实体个数(zset_max_ziplist_entries) > 0 
 // 并且 服务器配置里元素的长度(zset_max_ziplist_value) >= sdslen(c->argv[scoreidx+1]->ptr) 时
 // 创建压缩列表对象, 底层数据结构是ziplist 
 zobj = createZsetZiplistObject();
 }
 dbAdd(c->db,key,zobj);
 } else {
 if (zobj->type != OBJ_ZSET) {
 addReply(c,shared.wrongtypeerr);
 goto cleanup;
 }
 }
 // 往有序集合添加元素逻辑
 for (j = 0; j < elements; j++) {
 double newscore;
 score = scores[j];
 int retflags = flags;
 ele = c->argv[scoreidx+1+j*2]->ptr;
 // 真正往有序集合添加元素添加逻辑
 int retval = zsetAdd(zobj, score, ele, &retflags, &newscore);
 if (retval == 0) {
 addReplyError(c,nanerr);
 goto cleanup;
 }
 if (retflags & ZADD_ADDED) added++;
 if (retflags & ZADD_UPDATED) updated++;
 if (!(retflags & ZADD_NOP)) processed++;
 score = newscore;
 }
 server.dirty += (added+updated);
reply_to_client:
 if (incr) { /* ZINCRBY or INCR option. */
 if (processed)
 addReplyDouble(c,score);
 else
 addReplyNull(c);
 } else { /* ZADD. */
 addReplyLongLong(c,ch ? added+updated : added);
 }
cleanup:
 zfree(scores);
 if (added || updated) {
 signalModifiedKey(c,c->db,key);
 notifyKeyspaceEvent(NOTIFY_ZSET,
 incr ? "zincr" : "zadd", key, c->db->id);
 }
}

参考资料

[1] Redis源码剖析与实战 - 05 | 有序集合为何能同时支持点查询和范围查询?
[2] redis为什么选择了跳跃表而不是红黑树

Redis源码剖析与实战 学习笔记 Day5 05 | 有序集合为何能同时支持点查询和范围查询?
https://time.geekbang.org/col...

作者:wkq2786130原文地址:https://segmentfault.com/a/1190000043350807

%s 个评论

要回复文章请先登录注册