Redis原始碼分析之set 和 sorted set 使用

# 向集合新增一個或多個成員
SADD key member1 [member2]

# 獲取集合的成員數
SCARD key
# 返回第一個集合與其他集合之間的差異。
SDIFF key1 [key2]
# 返回給定所有集合的差集並儲存在 destination 中
SDIFFSTORE destination key1 [key2]
# 返回給定所有集合的交集
SINTER key1 [key2]
# 返回給定所有集合的交集並儲存在 destination 中
SINTERSTORE destination key1 [key2]
# 判斷 member 元素是否是集合 key 的成員
SISMEMBER key member
# 返回集合中的所有成員
SMEMBERS key
# 將 member 元素從 source 集合移動到 destination 集合
SMOVE source destination member
# 移除並返回集合中的一個隨機元素
SPOP key
# 返回集合中一個或多個亂數
SRANDMEMBER key [count]
# 移除集合中一個或多個成員
SREM key member1 [member2]
# 返回所有給定集合的並集
SUNION key1 [key2]
# 所有給定集合的並集儲存在 destination 集合中
SUNIONSTORE destination key1 [key2]
# 迭代集合中的元素
SSCAN key cursor [MATCH pattern] [COUNT count]

127.0.0.1:6379>  SADD set-test xiaoming
(integer) 1
(integer) 0
127.0.0.1:6379>  SADD set-test xiaoming1
127.0.0.1:6379>  SADD set-test xiaoming2

127.0.0.1:6379> SMEMBERS set-test
1) "xiaoming2"
2) "xiaoming"
3) "xiaoming1"

void saddCommand(client *c) {
    robj *set;
    int j, added = 0;

    set = lookupKeyWrite(c->db,c->argv[1]);
    if (checkType(c,set,OBJ_SET)) return;
    
    if (set == NULL) {
        set = setTypeCreate(c->argv[2]->ptr);
        dbAdd(c->db,c->argv[1],set);
    }
    for (j = 2; j < c->argc; j++) {
        if (setTypeAdd(set,c->argv[j]->ptr)) added++;
    if (added) {
        signalModifiedKey(c,c->db,c->argv[1]);
        notifyKeyspaceEvent(NOTIFY_SET,"sadd",c->argv[1],c->db->id);
    server.dirty += added;
    addReplyLongLong(c,added);
}
// 當 value 是整數型別使用 intset
// 否則使用雜湊表
robj *setTypeCreate(sds value) {
    if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)
        return createIntsetObject();
    return createSetObject();
/* Add the specified value into a set.
 *
 * If the value was already member of the set, nothing is done and 0 is
 * returned, otherwise the new element is added and 1 is returned. */
int setTypeAdd(robj *subject, sds value) {
    long long llval;
    // 如果是 OBJ_ENCODING_HT 說明是雜湊型別
    if (subject->encoding == OBJ_ENCODING_HT) {
        dict *ht = subject->ptr;
        dictEntry *de = dictAddRaw(ht,value,NULL);
        if (de) {
            // 設定key ,value 設定成null
            dictSetKey(ht,de,sdsdup(value));
            dictSetVal(ht,de,NULL);
            return 1;
        }
    // OBJ_ENCODING_INTSET 代表是 inset
    } else if (subject->encoding == OBJ_ENCODING_INTSET) {
        if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
            uint8_t success = 0;
            subject->ptr = intsetAdd(subject->ptr,llval,&success);
            if (success) {
                /* Convert to regular set when the intset contains
                 * too many entries. */
                // 如果條目過多將會轉換成集合
                size_t max_entries = server.set_max_intset_entries;
                /* limit to 1G entries due to intset internals. */
                if (max_entries >= 1<<30) max_entries = 1<<30;
                if (intsetLen(subject->ptr) > max_entries)
                    setTypeConvert(subject,OBJ_ENCODING_HT);
                return 1;
            }
        } else {
            /* Failed to get integer from object, convert to regular set. */
            setTypeConvert(subject,OBJ_ENCODING_HT);
            /* The set *was* an intset and this value is not integer
             * encodable, so dictAdd should always work. */
            serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK);
    } else {
        serverPanic("Unknown set encoding");
    return 0;

typedef struct intset {
    // 編碼方法，指定當前儲存的是 16 位，32 位，還是 64 位的整數
    uint32_t encoding;
    uint32_t length;
    // 實際儲存元素的陣列
    int8_t contents[];
} intset;

/* Insert an integer in the intset */
intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
    // 計算value的編碼長度
    uint8_t valenc = _intsetValueEncoding(value);
    uint32_t pos;
    if (success) *success = 1;

    /* Upgrade encoding if necessary. If we need to upgrade, we know that
     * this value should be either appended (if > 0) or prepended (if < 0),
     * because it lies outside the range of existing values. */
    // 如果value的編碼長度大於當前的編碼位數，進行升級
    if (valenc > intrev32ifbe(is->encoding)) {
        /* This always succeeds, so we don't need to curry *success. */
        return intsetUpgradeAndAdd(is,value);
    } else {
        /* Abort if the value is already present in the set.
         * This call will populate "pos" with the right position to insert
         * the value when it cannot be found. */
        // 當前值不存在的時候才進行插入  
        if (intsetSearch(is,value,&pos)) {
            if (success) *success = 0;
            return is;
        }
        is = intsetResize(is,intrev32ifbe(is->length)+1);
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
    }
    // 資料插入
    _intsetSet(is,pos,value);
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

/* Search for the position of "value". Return 1 when the value was found and
 * sets "pos" to the position of the value within the intset. Return 0 when
 * the value is not present in the intset and sets "pos" to the position
 * where "value" can be inserted. */
// 如果找到了對應的資料，返回 1 將 pos 設定為對應的位置
// 如果找不到，返回0，設定 pos 為可以為資料可以插入的位置
// intset 中的資料是排好序的，所以使用二分查詢來尋找對應的元素  
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
    int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;
    int64_t cur = -1;

    /* The value can never be found when the set is empty */
    if (intrev32ifbe(is->length) == 0) {
        if (pos) *pos = 0;
        return 0;
    } else {
        /* Check for the case where we know we cannot find the value,
         * but do know the insert position. */
        if (value > _intsetGet(is,max)) {
            if (pos) *pos = intrev32ifbe(is->length);
            return 0;
        } else if (value < _intsetGet(is,0)) {
            if (pos) *pos = 0;
        }
    }
    // 使用二分查詢
    while(max >= min) {
        mid = ((unsigned int)min + (unsigned int)max) >> 1;
        cur = _intsetGet(is,mid);
        if (value > cur) {
            min = mid+1;
        } else if (value < cur) {
            max = mid-1;
        } else {
            break;
    if (value == cur) {
        if (pos) *pos = mid;
        return 1;
        if (pos) *pos = min;
}

typedef struct dict {
    dictType *type;
    void *privdata;
    // 雜湊表陣列，長度為2，一個正常儲存資料，一個用來擴容
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
} dict;

// 雜湊表結構，通過兩個雜湊表使用，實現增量的 rehash  
typedef struct dictht {
    dictEntry **table;
    // hash 容量大小
    unsigned long size;
    // 總是等於 size - 1，用於計算索引值
    unsigned long sizemask;
    // 實際儲存的 dictEntry 數量
    unsigned long used;
} dictht;
//  k/v 鍵值對節點，是實際儲存資料的節點  
typedef struct dictEntry {
    // 鍵物件，總是一個字串型別的物件
    void *key;
    union {
        // void指標，這意味著它可以指向任何型別
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    // 指向下一個節點
    struct dictEntry *next;
} dictEntry;

dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing)
{
    long index;
    dictEntry *entry;
    dictht *ht;

    if (dictIsRehashing(d)) _dictRehashStep(d);
    /* Get the index of the new element, or -1 if
     * the element already exists. */
    if ((index = _dictKeyIndex(d, key, dictHashKey(d,key), existing)) == -1)
        return NULL;
    /* Allocate the memory and store the new entry.
     * Insert the element in top, with the assumption that in a database
     * system it is more likely that recently added entries are accessed
     * more frequently. */
    // 這裡來判斷是否正在 Rehash 中
    ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0];
    entry = zmalloc(sizeof(*entry));
    entry->next = ht->table[index];
    ht->table[index] = entry;
    ht->used++;
    /* Set the hash entry fields. */
    // 插入具體的資料  
    dictSetKey(d, entry, key);
    return entry;
}

int dictRehash(dict *d, int n) {
    int empty_visits = n*10; /* Max number of empty buckets to visit. */
    // 這裡來判斷是否正在 Rehash 中
    if (!dictIsRehashing(d)) return 0;

    // used 實際儲存的 dictEntry 數量
    // 如果有資料進行下面的遷移
    while(n-- && d->ht[0].used != 0) {
        dictEntry *de, *nextde;
        /* Note that rehashidx can't overflow as we are sure there are more
         * elements because ht[0].used != 0 */
        assert(d->ht[0].size > (unsigned long)d->rehashidx);
        while(d->ht[0].table[d->rehashidx] == NULL) {
            d->rehashidx++;
            if (--empty_visits == 0) return 1;
        }
        // 獲取老資料中的資料
        de = d->ht[0].table[d->rehashidx];
        /* Move all the keys in this bucket from the old to the new hash HT */
        while(de) {
            uint64_t h;
            nextde = de->next;
            /* Get the index in the new hash table */
            // 獲取新雜湊表的索引
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            de->next = d->ht[1].table[h];
            // 將資料放入到新的 dity 中
            d->ht[1].table[h] = de;
            d->ht[0].used--;
            d->ht[1].used++;
            de = nextde;
        d->ht[0].table[d->rehashidx] = NULL;
        d->rehashidx++;
    }
    /* Check if we already rehashed the whole table... */
    // 這裡來檢測是否完成了整個的 rehash 操作  
    if (d->ht[0].used == 0) {
        zfree(d->ht[0].table);
        // 將 ht[1] 的內容放入到 ht[0] 中，ht[1] 作為備用，下次 rehash 使用
        d->ht[0] = d->ht[1];
        _dictReset(&d->ht[1]);
        d->rehashidx = -1;
        return 0;
    /* More to rehash... */
    return 1;
}
// 使用增量的方式 rehash 
// 當然資料很大的話，一次遷移所有的資料，顯然是不合理的，會造成Redis執行緒阻塞，無法服務其他請求。這裡 Redis 使用的是漸進式 rehash。
// 在 rehash 期間，每次執行新增，刪除，查詢或者更新操作時，除了對命令本身的處理，還會順帶將雜湊表1中的資料拷貝到雜湊表2中。從索引0開始，每執行一次操作命令，拷貝一個索引位置的資料。  
// 如果沒有讀入和寫入操作，這時候就不能進行 rehash
// 所以會定時執行一定數量的 rehash 操作  
int incrementallyRehash(int dbid) {
    /* Keys dictionary */
    if (dictIsRehashing(server.db[dbid].dict)) {
        dictRehashMilliseconds(server.db[dbid].dict,1);
        return 1; /* already used our millisecond for this loop... */
    /* Expires */
    if (dictIsRehashing(server.db[dbid].expires)) {
        dictRehashMilliseconds(server.db[dbid].expires,1);
    return 0;

# 向有序集合新增一個或多個成員，或者更新已存在成員的分數
ZADD key score1 member1 [score2 member2]

# 獲取有序集合的成員數
ZCARD key
# 計算在有序集合中指定區間分數的成員數
ZCOUNT key min max
# 有序集合中對指定成員的分數加上增量 increment
ZINCRBY key increment member
# 計算給定的一個或多個有序集的交集並將結果集儲存在新的有序集合 destination 中
ZINTERSTORE destination numkeys key [key ...]
# 在有序集合中計算指定字典區間內成員數量
ZLEXCOUNT key min max
# 通過索引區間返回有序集合指定區間內的成員
ZRANGE key start stop [WITHSCORES]
# 通過字典區間返回有序集合的成員
ZRANGEBYLEX key min max [LIMIT offset count]
# 通過分數返回有序集合指定區間內的成員
ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT]
# 返回有序集合中指定成員的索引
ZRANK key member
# 移除有序集合中的一個或多個成員
ZREM key member [member ...]
# 移除有序集合中給定的字典區間的所有成員
ZREMRANGEBYLEX key min max
# 移除有序集合中給定的排名區間的所有成員
ZREMRANGEBYRANK key start stop
# 移除有序集合中給定的分數區間的所有成員
ZREMRANGEBYSCORE key min max
# 返回有序集中指定區間內的成員，通過索引，分數從高到低
ZREVRANGE key start stop [WITHSCORES]
# 返回有序集中指定分數區間內的成員，分數從高到低排序
ZREVRANGEBYSCORE key max min [WITHSCORES]
# 返回有序集合中指定成員的排名，有序整合員按分數值遞減(從大到小)排序
ZREVRANK key member
# 返回有序集中，成員的分數值
ZSCORE key member
# 計算給定的一個或多個有序集的並集，並儲存在新的 key 中
ZUNIONSTORE destination numkeys key [key ...]
# 迭代有序集合中的元素（包括元素成員和元素分值）
ZSCAN key cursor [MATCH pattern] [COUNT count]

127.0.0.1:6379> ZADD test-sset 1 member1
(integer) 1
127.0.0.1:6379> ZADD test-sset 2 member2
(integer) 1
127.0.0.1:6379> ZADD test-sset 3 member3
(integer) 1
127.0.0.1:6379> ZADD test-sset 3 member3
(integer) 0
127.0.0.1:6379> ZADD test-sset 4 member3
(integer) 0
127.0.0.1:6379> ZADD test-sset 5 member5
(integer) 1
127.0.0.1:6379> ZRANGE test-sset 0 10 WITHSCORES
1) "member1"
2) "1"
3) "member2"
4) "2"
5) "member3"
6) "4"
7) "member5"
8) "5"

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

typedef struct zskiplist {
    // 頭，尾節點
    struct zskiplistNode *header, *tail;
    // 節點數量
    unsigned long length;
    // 目前表內節點的最大層數
    int level;
} zskiplist;
/* ZSETs use a specialized version of Skiplists */
// ZSETs 使用的是特定版的 Skiplists
typedef struct zskiplistNode {
    // 這裡使用 sds 儲存具體的元素
    sds ele;
    // 元素的權重
    double score;
    // 後向指標（為了便於從跳錶的尾節點倒序查詢）
    struct zskiplistNode *backward;
    // 層
    // 儲存著指向其他元素的指標。高層的指標越過的元素數量大於等於低層的指標，為了提高查詢的效率，程式總是從高層先開始存取，然後隨著元素值範圍的縮小，慢慢降低層次。
    struct zskiplistLevel {
        // 節點上的前向指標
        struct zskiplistNode *forward;
        // 跨度
        // 用於記錄兩個節點之間的距離
        unsigned long span;
    } level[];
} zskiplistNode;

// 程式碼地址 https://github.com/redis/redis/blob/6.2/src/t_zset.c
// ZADD 命令
void zaddCommand(client *c) {
    zaddGenericCommand(c,ZADD_IN_NONE);
}

/* This generic command implements both ZADD and ZINCRBY. */
void zaddGenericCommand(client *c, int flags) {
    ...
    /* Lookup the key and create the sorted set if does not exist. */
    zobj = lookupKeyWrite(c->db,key);
    if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
    if (zobj == NULL) {
        if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
        // 超過閾值（zset-max-ziplist-entries、zset-max-ziplist-value）後，使用 hashtable + skiplist 儲存
        if (server.zset_max_ziplist_entries == 0 ||
            server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
        {
            zobj = createZsetObject();
        } else {
            zobj = createZsetZiplistObject();
        }
        dbAdd(c->db,key,zobj);
    }
// 看下具體的插入過程
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;
    /* Update the sorted set according to its encoding. */
    // 如果型別是 ZIPLIST
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *eptr;
        if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
             ...
            /* Remove and re-insert when score changed. */
            // 當分數改變時移除並重新插入
            if (score != curscore) {
                zobj->ptr = zzlDelete(zobj->ptr,eptr);
                zobj->ptr = zzlInsert(zobj->ptr,ele,score);
                *out_flags |= ZADD_OUT_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 {
                if (newscore) *newscore = score;
                *out_flags |= ZADD_OUT_ADDED;
                return 1;
            *out_flags |= ZADD_OUT_NOP;
    /* 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;
        // 在雜湊表中查詢元素
        de = dictFind(zs->dict,ele);
        if (de != NULL) {
            /* NX? Return, same element already exists. */
            if (nx) {
                *out_flags |= ZADD_OUT_NOP;
            // 雜湊表中獲取元素的權重
            curscore = *(double*)dictGetVal(de);
            /* 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)) {
            if (newscore) *newscore = score;
            /* Remove and re-insert when score changes. */
            // 如果權重發生變化了
                znode = zslUpdateScore(zs->zsl,curscore,ele,score);
                /* Note that we did not removed the original element from
                 * the hash table representing the sorted set, so we just
                 * update the score. */
                dictGetVal(de) = &znode->score; /* Update score ptr. */
        // 如果新元素不存在
            ele = sdsdup(ele);
            // 插入到跳錶節點
            znode = zslInsert(zs->zsl,score,ele);
            // 在雜湊表中插入
            serverAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
            *out_flags |= ZADD_OUT_ADDED;
    } else {
        serverPanic("Unknown sorted set encoding");
    return 0; /* Never reached. */

// 獲取有序集合中, 指定資料的排名.
// 若reverse==0, 排名以得分升序排列. 否則排名以得分降序排列.
// 第一個資料的排名為0, 而不是1
// 使用壓縮列表或者跳錶，裡面的資料都是排好序的
long zsetRank(robj *zobj, sds ele, int reverse) {
    unsigned long llen;
    unsigned long rank;

    llen = zsetLength(zobj);
    // 壓縮列表
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *zl = zobj->ptr;
        unsigned char *eptr, *sptr;
        eptr = ziplistIndex(zl,0);
        serverAssert(eptr != NULL);
        sptr = ziplistNext(zl,eptr);
        serverAssert(sptr != NULL);
        rank = 1;
        while(eptr != NULL) {
            if (ziplistCompare(eptr,(unsigned char*)ele,sdslen(ele)))
                break;
            rank++;
            zzlNext(zl,&eptr,&sptr);
        }
        if (eptr != NULL) {
             // 逆向取rank
             // 返回後面的資料
            if (reverse)
                return llen-rank;
            else
                return rank-1;
        } else {
            return -1;
    // 跳錶
    } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
        zset *zs = zobj->ptr;
        zskiplist *zsl = zs->zsl;
        dictEntry *de;
        double score;
        de = dictFind(zs->dict,ele);
        if (de != NULL) {
            score = *(double*)dictGetVal(de);
            // 查詢包含給定分值和成員物件的節點在跳躍表中的排位
            rank = zslGetRank(zsl,score,ele);
            /* Existing elements always have a rank. */
            serverAssert(rank != 0);
            // 逆向取rank
    } else {
        serverPanic("Unknown sorted set encoding");
    }
}
/*
 * 查詢包含給定分值和成員物件的節點在跳躍表中的排位。
 *
 * 如果沒有包含給定分值和成員物件的節點，返回 0 ，否則返回排位。
 * 注意，因為跳躍表的表頭也被計算在內，所以返回的排位以 1 為起始值。
 * T_wrost = O(N), T_avg = O(log N)
 */
unsigned long zslGetRank(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *x;
    unsigned long rank = 0;
    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))) {
            rank += x->level[i].span;
            // 沿著前進指標遍歷跳躍表
            x = x->level[i].forward;
        /* x might be equal to zsl->header, so test if obj is non-NULL */
        if (x->ele && sdscmp(x->ele,ele) == 0) {
            return rank;
    return 0;