有序汇合(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) 1127.0.0.1:6379> zadd zset_user_1 2 redis(integer) 1127.0.0.1:6379> zadd zset_user_1 3 es(integer) 1127.0.0.1:6379> zadd zset_user_1 4 hbase(integer) 1127.0.0.1:6379> zadd zset_user_1 5 clickhouse(integer) 1127.0.0.1:6379>127.0.0.1:6379> zrange zset_user_1 0 101) "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 41) "redis"2) "es"3) "hbase"127.0.0.1:6379>127.0.0.1:6379> zrangebyscore zset_user_1 2 4 withscores1) "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...