13、Redis 源码解析 - 有序集合类型键实现(t_zset)

有序集合类型键实现

1. 有序集合命令

2. 有序集合类型实现

有序集合对象的底层实现类型如下表：

关于底层的数据结构剖析和实现，请看如下博文

Redis 跳跃表源码剖析和注释

Redis 字典结构源码剖析和注释

Redis 压缩列表源码剖析和注释

一个有序集合对象的结构如下：

typedef struct redisObject {
    //对象的数据类型：OBJ_ZSET
    unsigned type:4;        
    //对象的编码类型：OBJ_ENCODING_SKIPLIST 或 OBJ_ENCODING_ZIPLIST
    unsigned encoding:4;

    //least recently used
    //实用LRU算法计算相对server.lruclock的LRU时间
    unsigned lru:LRU_BITS; /* lru time (relative to server.lruclock) */

    //引用计数
    int refcount;

    //指向底层数据实现的指针，指向zset结构或ziplist结构
    void *ptr;
} robj;

之前提到，编码为 OBJ_ENCODING_SKIPLIST 的底层的数据结构是 跳跃表和字典。因此就抽象了一个zset结构用来保存这两种结构。

typedef struct zset {
    dict *dict;         //字典
    zskiplist *zsl;     //跳跃表
} zset; //有序集合跳跃表类型

Redis采用跳跃表和哈希表代替平衡树的原因：具体分析可看 Redis 跳跃表源码剖析

• 跳跃表和平衡树的元素都是有序排列，而哈希表不是有序的。因此在哈希表上的查找只能是单个key的查找，不适合做范围查找。
• 跳跃表和平衡树做范围查找时，跳跃表算法简单，实现方便，而平衡树逻辑复杂。
• 查找单个key，跳跃表和平衡树的平均时间复杂度都为O(logN)，而哈希表的时间复杂度为O(N)。
• 跳跃表平均每个节点包含1.33个指针，而平衡树每个节点包含2个指针，更加节约内存。

因此，在redis中实现有序集合的办法是：跳跃表+哈希表

1、 跳跃表元素有序，而且可以范围查找，且实现起来比平衡树简单；
2、 哈希表查找单个key时间复杂度性能高；

我们创建一个视频播放量的有序集合，有体育sport、卡通carton、电视剧telepaly这三个成员

127.0.0.1:6379> ZADD video:view 1234 sport 888 carton 2345 teleplay 
(integer) 3
127.0.0.1:6379> ZRANGE video:view 0 -1 WITHSCORES
1) "carton"
2) "88888"
3) "sport"
4) "123456"
5) "teleplay"
6) "234567"
127.0.0.1:6379> OBJECT ENCODING video:view
"ziplist"

我们可以查看到编码类型是压缩列表，这个 video:view 对象的空间结构如下图所示：

当有序集合类型达到配置文件中的阈值条件时，就会发生编码类型的转换，转换为OBJ_ENCODING_SKIPLIST类型，而且转换是双向的，可以相互转换，默认的条件如下

#define OBJ_ZSET_MAX_ZIPLIST_ENTRIES 128    // ziplist中最多存放的节点数
#define OBJ_ZSET_MAX_ZIPLIST_VALUE 64       // ziplist中最大存放的数据长度

为了简单，我们就用上面的video:view 对象，使用OBJ_ENCODING_SKIPLIST编码方式展示可能的空间结构：

• 红色的是成员，我特地用红色表明，是想说明：OBJ_ENCODING_SKIPLIST 编码方式的有序集合，他的成员在两种数据结构中是共享的，只是简单的增加了引用计数，实际上内存只有一份成员数据。

有序集合基于ziplist的接口封装了一套对于OBJ_ENCODING_SKIPLIST类型的有序集合的API，这些函数大都是zzl开头：Redis有序集合键源码注释

// 从sptr指向的entry中取出有序集合的分值
double zzlGetScore(unsigned char *sptr);
// 返回一个sptr指向entry所构建的字符串类型的对象地址
robj *ziplistGetObject(unsigned char *sptr); 
// 比较eptr和cstr指向的元素，返回0表示相等，正整数表示eptr比cstr大
int zzlCompareElements(unsigned char *eptr, unsigned char *cstr, unsigned int clen);
// 返回有序集合的元素个数
unsigned int zzlLength(unsigned char *zl);
// 将当前的元素指针eptr和当前元素分值的指针sptr都指向下一个元素和元素的分值
void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
// 将当前的元素指针eptr和当前元素分值的指针sptr都指向上一个元素和元素的分值
void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
// 判断ziplist中是否有entry节点在range范围之内，有一个则返回1，否则返回0
int zzlIsInRange(unsigned char *zl, zrangespec *range);
// 返回第一个满足range范围的entry节点的地址
unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec *range);
// 返回最后一个满足range范围的entry节点的地址
unsigned char *zzlLastInRange(unsigned char *zl, zrangespec *range);
// 比较p指向的entry的值和规定范围spec，返回1表示大于spec的最大值
static int zzlLexValueGteMin(unsigned char *p, zlexrangespec *spec);
// 比较p指向的entry的值和规定范围spec，返回1表示小于spec的最小值
static int zzlLexValueLteMax(unsigned char *p, zlexrangespec *spec);
// 判断ziplist中是否有entry节点在range范围之内，有一个则返回1，否则返回0
int zzlIsInLexRange(unsigned char *zl, zlexrangespec *range);
// 返回第一个满足range范围的元素的地址
unsigned char *zzlFirstInLexRange(unsigned char *zl, zlexrangespec *range);
// 返回最后一个满足range范围的元素的地址
unsigned char *zzlLastInLexRange(unsigned char *zl, zlexrangespec *range); 
// 从ziplist中查找ele，将分值保存在score中
unsigned char *zzlFind(unsigned char *zl, robj *ele, double *score); 
// 删除eptr制定的元素和分值从ziplist中
unsigned char *zzlDelete(unsigned char *zl, unsigned char *eptr); 
//将ele元素和分值score插入到eptr指向节点的前面
unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, double score);
//将ele元素和分值score插入在ziplist中，从小到大排序
unsigned char *zzlInsert(unsigned char *zl, robj *ele, double score); 
// 删除ziplist中分值在制定范围内的元素，并保存删除的数量在deleted中
unsigned char *zzlDeleteRangeByScore(unsigned char *zl, zrangespec *range, unsigned long *deleted) 
// 删除ziplist中分值在制定字典序范围内的元素，并保存删除的数量在deleted中
unsigned char *zzlDeleteRangeByLex(unsigned char *zl, zlexrangespec *range, unsigned long *deleted);
// 删除给定下标范围内的所有元素
unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsigned int end, unsigned long *deleted);

这些API都是基于ziplist的结构封装，没有直接使用ziplist的接口原因是：因为元素和分值都是成对保存在ziplist的节点中，基于ziplist接口封装，是为了更简答的操作有序集合类型的数据。

而OBJ_ENCODING_SKIPLIST类型的有序集合，Redis则是直接使用跳跃表和字典所提供的接口。

因此，有序集合类型的API如下：Redis有序集合键源码注释

// 返回有序集合的元素个数
unsigned int zsetLength(robj *zobj)
// 将有序集合对象的编码转换为encoding制定的编码类型
void zsetConvert(robj *zobj, int encoding) 
// 按需转换编码成OBJ_ENCODING_ZIPLIST
void zsetConvertToZiplistIfNeeded(robj *zobj, size_t maxelelen)
// 将有序集合的member成员的分值保存到score中
int zsetScore(robj *zobj, robj *member, double *score)

3. 有序集合的迭代器

有序集合在t_zset.c中定义了迭代器，迭代器适用于有序集合和集合。

// 集合类型的迭代器
typedef struct {
    robj *subject;                  //所属的对象
    int type; /* Set, sorted set */ //对象的类型，可以是集合或有序集合
    int encoding;                   //编码
    double weight;                  //权重

    union {
        /* Set iterators. */        //集合迭代器
        union _iterset {
            struct {                //intset迭代器
                intset *is;         //所属的intset
                int ii;             //intset节点下标
            } is;
            struct {                //字典迭代器
                dict *dict;         //所属的字典
                dictIterator *di;   //字典的迭代器
                dictEntry *de;      //当前指向的字典节点
            } ht;
        } set;

        /* Sorted set iterators. */         //有序集合迭代器
        union _iterzset {                   //ziplist迭代器
            struct {
                unsigned char *zl;          //所属的ziplist
                unsigned char *eptr, *sptr; //当前指向的元素和分值
            } zl;
            struct {                        //zset迭代器
                zset *zs;                   //所属的zset
                zskiplistNode *node;        //当前指向的跳跃节点
            } sl;
        } zset;
    } iter;
} zsetopsrc;
// 集合迭代器和有序集合迭代器是一个共用体union，因此，不会浪费空间。

在实现迭代的过程中，由于迭代器指向的节点可能数据类型各不相同，有对象、有字符串、有整数，因此Redis又定义了一个保存这些类型的统一结构：

/* Store value retrieved from the iterator. */
// 从集合类型迭代器中恢复存储的值
typedef struct {
    int flags;
    unsigned char _buf[32]; /* Private buffer. */
    // 保存元素的各种类型
    robj *ele;              //对象
    unsigned char *estr;    //ziplist
    unsigned int elen;
    long long ell;
    // 保存分值
    double score;
} zsetopval;

定义了zsetopval结构，还可以实现内部各种类型的数据进行转换，例如：根据estr的值，转换成ele对象类型。

当然，还为迭代器定义了一些接口，使用起来方便，函数名字都是以zui开头：Redis有序集合键源码注释

// 初始化迭代器
void zuiInitIterator(zsetopsrc *op)
// 释放迭代器空间
void zuiClearIterator(zsetopsrc *op)
// 返回正在被迭代的元素的长度
int zuiLength(zsetopsrc *op)
// 将当前迭代器指向的元素保存到val中，并更新指向，指向下一个元素
int zuiNext(zsetopsrc *op, zsetopval *val)
// 根据val设置longlong值
int zuiLongLongFromValue(zsetopval *val)
// 根据val中的值，创建对象
robj *zuiObjectFromValue(zsetopval *val)
// 从val中取出字符串
int zuiBufferFromValue(zsetopval *val)
// 在迭代器制定的对象中查找给定的元素，找到返回1，没找到返回0
int zuiFind(zsetopsrc *op, zsetopval *val, double *score)
// 对比两个被迭代对象的元素的个数，返回差值
int zuiCompareByCardinality(const void *s1, const void *s2) 

4. 有序集合的范围限定方式

Redis有序集合有三种范围限定

• 根据字典区间
• 根据排位区间
• 根据分值区间

以上三种主要依赖两种排序方式

• 分值序
• 字典序

以上两种排序所表示的范围，依赖于两个结构，一个规定分值序的范围和边界，一个规定字典序的范围和边界。

/* Struct to hold a inclusive/exclusive range spec by score comparison. */
// 分值序的范围和边界
typedef struct {
    double min, max;    //最小值和最大值

    //minex表示最小值是否包含在这个范围，1表示不包含，0表示包含
    //同理maxex表示最大值
    int minex, maxex; /* are min or max exclusive? */
} zrangespec;

/* Struct to hold an inclusive/exclusive range spec by lexicographic comparison. */
//  字典序的范围和边界
typedef struct {
    robj *min, *max;  /* May be set to shared.(minstring|maxstring) *///最小值和最大值

    //minex表示最小值是否包含在这个范围，1表示不包含，0表示包含
    //同理maxex表示最大值
    int minex, maxex; /* are min or max exclusive? */
} zlexrangespec;

这两种区别就是：zrangespec不需要释放，而zlexrangespec需要释放空间，因为字典序的最大值和最小值对象可能是共享的。

因为上面的两种区别，因此在有序集合命令实现时，也是非常麻烦，例如ZRANGEBYLEX和ZRANGEBYSCORE命令，由于排序范围和边界，都抽象成一个结构体，所以代码逻辑几乎一致，但是字典序的多了释放空间这一步，否则就可以用一个函数实现这两种不同排序。

5. 有序集合类型命令的实现

有序集合命令实现具体请看：Redis有序集合键源码注释 这里只剖析个别命令：Redis 有序集合命令详解

• ZUNIONSTORE ZINTERSTORE 命令的底层实现：

有序集合计算交集的一般算法：先将所有集合按元素个数从小到大排序，然后以元素个数最少的集合，也就是第一个集合为基数，遍历该集合的所有元素，如果元素在之后的所有集合都存在，则加入结果集中。

有序集合计算并集的一般算法：先将所有集合按元素个数从小到大排序，遍历所有的集合的所有元素，将元素添加到结果集中，如果元素已不存在，则添加，否则操作下一个元素。

#define REDIS_AGGR_SUM 1    //和
#define REDIS_AGGR_MIN 2    //最小数
#define REDIS_AGGR_MAX 3    //最大数
//默认使用的参数 SUM ，可以将所有集合中某个成员的 score 值之 和 作为结果集中该成员的 score 值；
//使用参数 MIN ，可以将所有集合中某个成员的 最小 score 值作为结果集中该成员的 score 值；
//而参数 MAX 则是将所有集合中某个成员的 最大 score 值作为结果集中该成员的 score 值。
// ZUNIONSTORE destination numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE SUM|MIN|MAX]
// ZINTERSTORE destination numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE SUM|MIN|MAX]
// ZUNIONSTORE ZINTERSTORE命令的底层实现
void zunionInterGenericCommand(client *c, robj *dstkey, int op) {
    int i, j;
    long setnum;
    int aggregate = REDIS_AGGR_SUM;     //默认是求和
    zsetopsrc *src;
    zsetopval zval;
    robj *tmp;
    unsigned int maxelelen = 0;         //判断是否需要编码转换的条件之一
    robj *dstobj;
    zset *dstzset;
    zskiplistNode *znode;
    int touched = 0;

    /* expect setnum input keys to be given */
    // 取出numkeys参数值，保存在setnum中
    if ((getLongFromObjectOrReply(c, c->argv[2], &setnum, NULL) != C_OK))
        return;

    // 至少指定一个key
    if (setnum < 1) {
        addReplyError(c,
            "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE");
        return;
    }

    /* test if the expected number of keys would overflow */
    // setnum与key数量不匹配，发送语法错误信息
    if (setnum > c->argc-3) {
        addReply(c,shared.syntaxerr);
        return;
    }

    /* read keys to be used for input */
    // 为每一个key分配一个迭代器空间
    src = zcalloc(sizeof(zsetopsrc) * setnum);
    // 遍历
    for (i = 0, j = 3; i < setnum; i++, j++) {
        // 以写方式读取出有序集合
        robj *obj = lookupKeyWrite(c->db,c->argv[j]);
        if (obj != NULL) {  //检查读取出的集合类型，否则发送信息
            if (obj->type != OBJ_ZSET && obj->type != OBJ_SET) {
                zfree(src);
                addReply(c,shared.wrongtypeerr);
                return;
            }

            // 初始化每个集合对象的迭代器
            src[i].subject = obj;
            src[i].type = obj->type;
            src[i].encoding = obj->encoding;
        } else {
            src[i].subject = NULL;  //集合对象不存在则设置为NULL
        }

        /* Default all weights to 1. */
        src[i].weight = 1.0;        //默认权重为1.0
    }

    /* parse optional extra arguments */
    // 解析其他的参数
    if (j < c->argc) {
        int remaining = c->argc - j;    //其他参数的数量

        while (remaining) {
            // 设置了权重WEIGHTS参数
            if (remaining >= (setnum + 1) && !strcasecmp(c->argv[j]->ptr,"weights")) {
                j++; remaining--;
                // 遍历设置的weight，保存到每个集合对象的迭代器中
                for (i = 0; i < setnum; i++, j++, remaining--) {
                    if (getDoubleFromObjectOrReply(c,c->argv[j],&src[i].weight,
                            "weight value is not a float") != C_OK)
                    {
                        zfree(src);
                        return;
                    }
                }
            // 设置了集合方式AGGREGATE参数
            } else if (remaining >= 2 && !strcasecmp(c->argv[j]->ptr,"aggregate")) {
                j++; remaining--;
                // 根据不同的参数，设置默认的方式
                if (!strcasecmp(c->argv[j]->ptr,"sum")) {
                    aggregate = REDIS_AGGR_SUM;
                } else if (!strcasecmp(c->argv[j]->ptr,"min")) {
                    aggregate = REDIS_AGGR_MIN;
                } else if (!strcasecmp(c->argv[j]->ptr,"max")) {
                    aggregate = REDIS_AGGR_MAX;
                } else {
                    zfree(src);
                    addReply(c,shared.syntaxerr);   //不是以上三种集合方式，发送语法错误信息
                    return;
                }
                j++; remaining--;
            } else {
                zfree(src);
                addReply(c,shared.syntaxerr);   //不是以上两种参数，发送语法错误信息
                return;
            }
        }
    }

    /* sort sets from the smallest to largest, this will improve our
     * algorithm's performance */
    // 对所有集合元素的多少，从小到大排序，提高算法性能
    qsort(src,setnum,sizeof(zsetopsrc),zuiCompareByCardinality);

    dstobj = createZsetObject();    //创建结果集对象
    dstzset = dstobj->ptr;
    memset(&zval, 0, sizeof(zval)); //初始化保存有序集合元素和分值的结构

    // ZINTERSTORE命令
    if (op == SET_OP_INTER) {
        /* Skip everything if the smallest input is empty. */
        // 如果当前集合没有元素，则不处理
        if (zuiLength(&src[0]) > 0) {
            /* Precondition: as src[0] is non-empty and the inputs are ordered
             * by size, all src[i > 0] are non-empty too. */
            // 初始化元素最少的有序集合
            zuiInitIterator(&src[0]);

            // 遍历元素最少的有序集合的每一个元素，将元素的分值和数值保存到zval中
            while (zuiNext(&src[0],&zval)) {
                double score, value;
                // 计算加权分值
                score = src[0].weight * zval.score;
                if (isnan(score)) score = 0;

                // 遍历其后的每一个有序集合，分别跟元素最少的有序集合做运算
                for (j = 1; j < setnum; j++) {
                    /* It is not safe to access the zset we are
                     * iterating, so explicitly check for equal object. */
                    // 如果输入的key一样
                    if (src[j].subject == src[0].subject) {
                        // 计算新的加权分值
                        value = zval.score*src[j].weight;
                        // 根据集合方式sum、min、max保存结果值到score
                        zunionInterAggregate(&score,value,aggregate);

                    // 输入的key与第一个key不一样，在当前key查找当前的元素，并将找到元素的分值保存在value中
                    } else if (zuiFind(&src[j],&zval,&value)) {
                        // 计算加权后的分值
                        value *= src[j].weight;
                        // 根据集合方式sum、min、max保存结果值到score
                        zunionInterAggregate(&score,value,aggregate);

                    // 如果当前元素在所有有序集合中没有找到，则跳过for循环，处理下一个元素
                    } else {
                        break;
                    }
                }

                /* Only continue when present in every input. */
                // 如果没有跳出for循环，那么该元素一定是每个key集合中都存在，因此进行ZINTER的STORE操作
                if (j == setnum) {
                    tmp = zuiObjectFromValue(&zval);    //创建当前元素的对象
                    znode = zslInsert(dstzset->zsl,score,tmp);  //插入到有序集合中
                    incrRefCount(tmp); /* added to skiplist */
                    dictAdd(dstzset->dict,tmp,&znode->score);   //加入到字典中
                    incrRefCount(tmp); /* added to dictionary */

                    if (sdsEncodedObject(tmp)) {    //如果是字符串编码的对象
                        if (sdslen(tmp->ptr) > maxelelen)   //更新最大长度
                            maxelelen = sdslen(tmp->ptr);
                    }
                }
            }
            zuiClearIterator(&src[0]);  //释放集合类型的迭代器
        }
    // ZUNIONSTORE命令
    } else if (op == SET_OP_UNION) {
        dict *accumulator = dictCreate(&setDictType,NULL);  //创建一个临时字典作为结果集字典
        dictIterator *di;
        dictEntry *de;
        double score;

        // 计算出最大有序集合元素个数，并集至少和最大的集合一样大，因此按需扩展结果集字典
        if (setnum) {
            /* Our union is at least as large as the largest set.
             * Resize the dictionary ASAP to avoid useless rehashing. */
            dictExpand(accumulator,zuiLength(&src[setnum-1]));
        }

        /* Step 1: Create a dictionary of elements -> aggregated-scores
         * by iterating one sorted set after the other. */
        // 1.遍历所有的集合
        for (i = 0; i < setnum; i++) {
            if (zuiLength(&src[i]) == 0) continue;  //跳过空集合

            zuiInitIterator(&src[i]);   //初始化集合迭代器
            // 遍历当前集合的所有元素，迭代器当前指向的元素保存在zval中
            while (zuiNext(&src[i],&zval)) {
                /* Initialize value */
                score = src[i].weight * zval.score;     //初始化分值
                if (isnan(score)) score = 0;

                /* Search for this element in the accumulating dictionary. */
                // 从结果集字典中找当前元素
                de = dictFind(accumulator,zuiObjectFromValue(&zval));
                /* If we don't have it, we need to create a new entry. */
                // 如果在结果集中找不到，则加入到结果集中
                if (de == NULL) {
                    tmp = zuiObjectFromValue(&zval);    //创建元素对象
                    /* Remember the longest single element encountered,
                     * to understand if it's possible to convert to ziplist
                     * at the end. */
                    if (sdsEncodedObject(tmp)) {    //字符串类型的话要更新maxelelen
                        if (sdslen(tmp->ptr) > maxelelen)
                            maxelelen = sdslen(tmp->ptr);
                    }
                    /* Add the element with its initial score. */
                    de = dictAddRaw(accumulator,tmp);   //加入到结果集字典中
                    incrRefCount(tmp);
                    dictSetDoubleVal(de,score);         //设置元素的分值

                // 如果在结果集中可以找到
                } else {
                    /* Update the score with the score of the new instance
                     * of the element found in the current sorted set.
                     *
                     * Here we access directly the dictEntry double
                     * value inside the union as it is a big speedup
                     * compared to using the getDouble/setDouble API. */
                    zunionInterAggregate(&de->v.d,score,aggregate); //更新加权分值
                }
            }
            zuiClearIterator(&src[i]);  //清理当前集合的迭代器
        }

        /* Step 2: convert the dictionary into the final sorted set. */
        // 2. 创建字典的迭代器，遍历结果集字典
        di = dictGetIterator(accumulator);

        /* We now are aware of the final size of the resulting sorted set,
         * let's resize the dictionary embedded inside the sorted set to the
         * right size, in order to save rehashing time. */
        // 按需扩展结果集合中的字典成员
        dictExpand(dstzset->dict,dictSize(accumulator));

        // 遍历结果集字典，
        while((de = dictNext(di)) != NULL) {
            // 获得元素对象和分值
            robj *ele = dictGetKey(de);
            score = dictGetDoubleVal(de);
            // 插入到结果集合的跳跃表中
            znode = zslInsert(dstzset->zsl,score,ele);
            incrRefCount(ele); /* added to skiplist */
            // 添加到结果集合的字典中
            dictAdd(dstzset->dict,ele,&znode->score);
            incrRefCount(ele); /* added to dictionary */
        }
        dictReleaseIterator(di);    //释放字典的迭代器

        /* We can free the accumulator dictionary now. */
        dictRelease(accumulator);   //释放临时保存结果的字典
    } else {
        serverPanic("Unknown operator");
    }

    // 删除已经存在的dstkey，并做标记
    if (dbDelete(c->db,dstkey))
        touched = 1;
    // 如果结果集元素不为0
    if (dstzset->zsl->length) {
        zsetConvertToZiplistIfNeeded(dstobj,maxelelen); //按需进行编码转换
        dbAdd(c->db,dstkey,dstobj);                     //将数据库中的dstkey和dstobj关联成对
        addReplyLongLong(c,zsetLength(dstobj));         //发送结果集合的元素数量给client
        signalModifiedKey(c->db,dstkey);                //发送键被修改的信号
        notifyKeyspaceEvent(NOTIFY_ZSET,                //发送对应的事件通知
            (op == SET_OP_UNION) ? "zunionstore" : "zinterstore",
            dstkey,c->db->id);
        server.dirty++;                                 //更新脏键
    // 结果集没有元素，释放对象，发送0给client
    } else {
        decrRefCount(dstobj);
        addReply(c,shared.czero);
        if (touched) {
            signalModifiedKey(c->db,dstkey);
            notifyKeyspaceEvent(NOTIFY_GENERIC,"del",dstkey,c->db->id);
            server.dirty++;
        }
    }
    zfree(src); //释放集合迭代器数组空间
}

• ZRANGEBYSCORE ZREVRANGEBYSCORE命令的底层实现

/* This command implements ZRANGEBYSCORE, ZREVRANGEBYSCORE. */
// ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]
// ZREVRANGEBYSCORE key max min [WITHSCORES] [LIMIT offset count]
// ZRANGEBYSCORE ZREVRANGEBYSCORE命令的底层实现
void genericZrangebyscoreCommand(client *c, int reverse) {
    zrangespec range;
    robj *key = c->argv[1];
    robj *zobj;
    long offset = 0, limit = -1;
    int withscores = 0;
    unsigned long rangelen = 0;
    void *replylen = NULL;
    int minidx, maxidx;

    /* Parse the range arguments. */
    // 解析min和max范围的参数的下标
    if (reverse) {
        /* Range is given as [max,min] */
        maxidx = 2; minidx = 3;
    } else {
        /* Range is given as [min,max] */
        minidx = 2; maxidx = 3;
    }

    // 解析min和max范围参数，保存到zrangespec中，默认是包含[]临界值
    if (zslParseRange(c->argv[minidx],c->argv[maxidx],&range) != C_OK) {
        addReplyError(c,"min or max is not a float");
        return;
    }

    /* Parse optional extra arguments. Note that ZCOUNT will exactly have
     * 4 arguments, so we'll never enter the following code path. */
    // 解析其他的参数
    if (c->argc > 4) {
        int remaining = c->argc - 4;    //其他参数个数
        int pos = 4;

        while (remaining) {
            // 如果有WITHSCORES，设置标志
            if (remaining >= 1 && !strcasecmp(c->argv[pos]->ptr,"withscores")) {
                pos++; remaining--;
                withscores = 1;
            // 如果有LIMIT，取出offset和count值 保存在offset和limit中
            } else if (remaining >= 3 && !strcasecmp(c->argv[pos]->ptr,"limit")) {
                if ((getLongFromObjectOrReply(c, c->argv[pos+1], &offset, NULL) != C_OK) ||
                    (getLongFromObjectOrReply(c, c->argv[pos+2], &limit, NULL) != C_OK)) return;
                pos += 3; remaining -= 3;
            } else {
                addReply(c,shared.syntaxerr);
                return;
            }
        }
    }

    /* Ok, lookup the key and get the range */
    // 以读操作取出有序集合对象
    if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL ||
        checkType(c,zobj,OBJ_ZSET)) return;

    // ziplist
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *zl = zobj->ptr;
        unsigned char *eptr, *sptr;
        unsigned char *vstr;
        unsigned int vlen;
        long long vlong;
        double score;

        /* If reversed, get the last node in range as starting point. */
        // 获得range范围内的起始元素节点地址
        if (reverse) {
            eptr = zzlLastInRange(zl,&range);
        } else {
            eptr = zzlFirstInRange(zl,&range);
        }

        /* No "first" element in the specified interval. */
        // 没有元素在range里，发送空回复
        if (eptr == NULL) {
            addReply(c, shared.emptymultibulk);
            return;
        }

        /* Get score pointer for the first element. */
        serverAssertWithInfo(c,zobj,eptr != NULL);
        // 分值节点地址
        sptr = ziplistNext(zl,eptr);

        /* We don't know in advance how many matching elements there are in the
         * list, so we push this object that will represent the multi-bulk
         * length in the output buffer, and will "fix" it later */
        // 回复长度
        replylen = addDeferredMultiBulkLength(c);

        /* If there is an offset, just traverse the number of elements without
         * checking the score because that is done in the next loop. */
        // 跳过offset设定的长度
        while (eptr && offset--) {
            if (reverse) {
                zzlPrev(zl,&eptr,&sptr);
            } else {
                zzlNext(zl,&eptr,&sptr);
            }
        }

        // 遍历所有符合范围的节点
        while (eptr && limit--) {
            score = zzlGetScore(sptr);  //获取分值

            /* Abort when the node is no longer in range. */
            // 检查分值是否符合范围内的分值
            if (reverse) {
                if (!zslValueGteMin(score,&range)) break;
            } else {
                if (!zslValueLteMax(score,&range)) break;
            }

            /* We know the element exists, so ziplistGet should always succeed */
            serverAssertWithInfo(c,zobj,ziplistGet(eptr,&vstr,&vlen,&vlong));   //保存当前元素的值到参数中

            rangelen++;
            // 不同元素类型，发送不同的类型的值给client
            if (vstr == NULL) {
                addReplyBulkLongLong(c,vlong);
            } else {
                addReplyBulkCBuffer(c,vstr,vlen);
            }

            if (withscores) {
                addReplyDouble(c,score);    //发送分值
            }

            /* Move to next node */
            // 指向下一个元素和分值节点
            if (reverse) {
                zzlPrev(zl,&eptr,&sptr);
            } else {
                zzlNext(zl,&eptr,&sptr);
            }
        }
    // 跳跃表
    } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
        zset *zs = zobj->ptr;
        zskiplist *zsl = zs->zsl;
        zskiplistNode *ln;

        /* If reversed, get the last node in range as starting point. */
        // 获得range范围内的起始节点地址
        if (reverse) {
            ln = zslLastInRange(zsl,&range);
        } else {
            ln = zslFirstInRange(zsl,&range);
        }

        /* No "first" element in the specified interval. */
        // 没有元素存在在制定的范围内
        if (ln == NULL) {
            addReply(c, shared.emptymultibulk);
            return;
        }

        /* We don't know in advance how many matching elements there are in the
         * list, so we push this object that will represent the multi-bulk
         * length in the output buffer, and will "fix" it later */
        // 回复长度
        replylen = addDeferredMultiBulkLength(c);

        /* If there is an offset, just traverse the number of elements without
         * checking the score because that is done in the next loop. */
        // 跳过offset个节点
        while (ln && offset--) {
            if (reverse) {
                ln = ln->backward;
            } else {
                ln = ln->level[0].forward;
            }
        }

        // 遍历所有符合范围的节点
        while (ln && limit--) {
            /* Abort when the node is no longer in range. */
            // 检查分值是否符合
            if (reverse) {
                if (!zslValueGteMin(ln->score,&range)) break;
            } else {
                if (!zslValueLteMax(ln->score,&range)) break;
            }

            rangelen++;
            addReplyBulk(c,ln->obj);    //发送元素

            if (withscores) {
                addReplyDouble(c,ln->score);    //发送分值
            }

            /* Move to next node */
            // 指向下一个节点
            if (reverse) {
                ln = ln->backward;
            } else {
                ln = ln->level[0].forward;
            }
        }
    } else {
        serverPanic("Unknown sorted set encoding");
    }

    if (withscores) {   //更新回复长度
        rangelen *= 2;
    }

    setDeferredMultiBulkLength(c, replylen, rangelen);  //发送范围的长度
}

• ZRANK ZREVRANK命令底层实现

// ZRANK key member
// ZREVRANK key member
// ZRANK ZREVRANK命令底层实现
void zrankGenericCommand(client *c, int reverse) {
    robj *key = c->argv[1];
    robj *ele = c->argv[2];
    robj *zobj;
    unsigned long llen;
    unsigned long rank;

    // 以读操作取出有序集合对象
    if ((zobj = lookupKeyReadOrReply(c,key,shared.nullbulk)) == NULL ||
        checkType(c,zobj,OBJ_ZSET)) return;
    llen = zsetLength(zobj);    //获得集合元素数量

    serverAssertWithInfo(c,ele,sdsEncodedObject(ele));  //member参数必须是字符串类型对象

    // ziplist
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *zl = zobj->ptr;
        unsigned char *eptr, *sptr;

        eptr = ziplistIndex(zl,0);  //元素节点地址
        serverAssertWithInfo(c,zobj,eptr != NULL);
        sptr = ziplistNext(zl,eptr);//分值节点地址
        serverAssertWithInfo(c,zobj,sptr != NULL);

        rank = 1;//排位
        while(eptr != NULL) {
            // 比较当前元素和member参数的大小
            if (ziplistCompare(eptr,ele->ptr,sdslen(ele->ptr)))
                break;
            rank++; //如果大于，排位加1
            zzlNext(zl,&eptr,&sptr);    //指向下个元素和分值
        }

        // 发送排位值
        if (eptr != NULL) {
            if (reverse)
                addReplyLongLong(c,llen-rank);
            else
                addReplyLongLong(c,rank-1);
        } else {
            addReply(c,shared.nullbulk);
        }
    // skiplist
    } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
        zset *zs = zobj->ptr;
        zskiplist *zsl = zs->zsl;
        dictEntry *de;
        double score;

        ele = c->argv[2];   //member参数
        de = dictFind(zs->dict,ele);    //查找保存member的节点地址
        if (de != NULL) {
            score = *(double*)dictGetVal(de);   //获得分值
            rank = zslGetRank(zsl,score,ele);   //计算排位
            serverAssertWithInfo(c,ele,rank); /* Existing elements always have a rank. */
            // 发送排位
            if (reverse)
                addReplyLongLong(c,llen-rank);
            else
                addReplyLongLong(c,rank-1);
        } else {
            addReply(c,shared.nullbulk);
        }
    } else {
        serverPanic("Unknown sorted set encoding");
    }
}