渐进式解析 Redis 源码 - 集合类型 t_set

阅读量

集合类型 的对象编码方式为 OBJ_ENCODING_INTSET 和 OBJ_ENCODING_HT，对应的 底层数据结构为 整数集合 intset 和 字典 dict

编码转换

当 集合条件 可以同时满足以下条件的时候，对象使用 intset 结构，否则就使用 字典结构 作为底层实现

  • 集合对象保存的所有元素都是 整数
  • 集合对象的元素个数不能大于 512（这个限制是由 set-max-intset-entries 参数控制，对应 server 全局变量为 server.set_max_intset_entries，默认 512）

如果原本使用的是 intset 编码，当上述两个条件 任意一个条件不能被满足的时候，就会执行转码操作，原本保存在 整数集合中的元素都会被转移到 字典里面并保存，对象编码也会从 intset 转换成 hashtable

而 hashtable 结构的数据 元素值为字典的键，值一律为 NULL

下面我们就 Redis 集合类型 一些典型的 API 进行解析~~~

源码解析

由于有些方法里的代码量比较大，我们这里按照 典型的代码片段进行解析，同志们可以根据文章提示的代码位置 和 代码里面的关键词 在源码中搜素，可能数据结构一些元素 看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

首先我们上面提到的编码类型转换函数为 setTypeConvert()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
/* t_set.c: line 235 */
void setTypeConvert(robj *setobj, int enc) {
    setTypeIterator *si;
    // 判断类型是否为 intset
    serverAssertWithInfo(NULL,setobj,setobj->type == OBJ_SET && setobj->encoding == OBJ_ENCODING_INTSET);
    // 判断提供的 enc 编码类型为 hashtale
    if (enc == OBJ_ENCODING_HT) {
        int64_t intele;
        // 新建字典结构
        dict *d = dictCreate(&setDictType,NULL);
        sds element;
        // 扩张 字典大小
        dictExpand(d,intsetLen(setobj->ptr));
        // 初始化 迭代器，遍历 intset，依次 将元素假如到 字典中
        si = setTypeInitIterator(setobj);
        while (setTypeNext(si,&element,&intele) != -1) {
            // 整数转换为 字符串元素
            element = sdsfromlonglong(intele);
            // 添加到字典里
            serverAssert(dictAdd(d,element,NULL) == DICT_OK);
        }
        // 释放迭代器
        setTypeReleaseIterator(si);
        // 设置编码类型
        setobj->encoding = OBJ_ENCODING_HT;
        // 释放 intset 结构
        zfree(setobj->ptr);
        // 重新指向
        setobj->ptr = d;
    } else {
        // 否则提示不支持其他类型
        serverPanic("Unsupported set conversion");
    }
}

集合添加命令 sadd 对应的函数为 saddCommand()，主要的逻辑函数为 setTypeAdd()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/* t_set.c: line 264 */
void saddCommand(client *c) {
    robj *set;
    int j, added = 0;
    // 查找 键值对 是否存在
    set = lookupKeyWrite(c->db,c->argv[1]);
    // 如果不存在则 新建
    if (set == NULL) {
        set = setTypeCreate(c->argv[2]->ptr);
        dbAdd(c->db,c->argv[1],set);
    } else {
        // 如果存在但是类型不是 SET集合类型，则报错
        if (set->type != OBJ_SET) {
            addReply(c,shared.wrongtypeerr);
            return;
        }
    }
    for (j = 2; j < c->argc; j++) {
        // 依次写入元素
        if (setTypeAdd(set,c->argv[j]->ptr)) added++;
    }
    // 如果至少添加了一个元素
    if (added) {
        // 发送通知
        signalModifiedKey(c->db,c->argv[1]);
        // 发送 sadd 事件通知
        notifyKeyspaceEvent(NOTIFY_SET,"sadd",c->argv[1],c->db->id);
    }
    server.dirty += added;
    addReplyLongLong(c,added);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
/* t_set.c: line 52 */
int setTypeAdd(robj *subject, sds value) {
    long long llval;
    // 如果编码类型为 hashtable字典 类型
    if (subject->encoding == OBJ_ENCODING_HT) {
        // 获取 hashtable 主体
        dict *ht = subject->ptr;
        // 添加 字典键值对，键为 元素，值为 NULL
        dictEntry *de = dictAddRaw(ht,value,NULL);
        // 如果是已经存在
        if (de) {
            // 重新设置  字典key 为 元素值，字典value 为 NULL
            dictSetKey(ht,de,sdsdup(value));
            dictSetVal(ht,de,NULL);
            return 1;
        }
    } else if (subject->encoding == OBJ_ENCODING_INTSET) {
    // 如果为 整数集合
        // 判断待添加数据是否符合 intset 的规定
        if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
            uint8_t success = 0;
            // 数字添加至 intset 集合
            subject->ptr = intsetAdd(subject->ptr,llval,&success);
            // 添加成功
            if (success) {
                // 判断长度是否大于了 intset 最大限制数
                if (intsetLen(subject->ptr) > server.set_max_intset_entries)
                    // 转码为 hashtable
                    setTypeConvert(subject,OBJ_ENCODING_HT);
                return 1;
            }
        } else {
            // 直接转码为 hashtable
            setTypeConvert(subject,OBJ_ENCODING_HT);
            // 调用字典添加函数
            serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK);
            return 1;
        }
    } else {
        // 不符合 intset 和 hashtable 之外的类型
        serverPanic("Unknown set encoding");
    }
    return 0;
}

类似 sadd 命令，另外一些 删除、查找 的函数逻辑总体来说还是比较简单的

不过有一些 差集、并集 的命令（sdiff、sdiffstore、sunion、sunionstore） 很值得探讨一下，这些命令的核心处理函数都为 sunionDiffGenericCommand()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
/* t_set.c: line 929 */
// 并集运算
#define SET_OP_UNION 0
// 差集运算
#define SET_OP_DIFF  1
/**
 * @param robj **setkeys 为需要对比的 key 的集合
 * @param int  setnum    为需要对比的 key 的个数
 * @param rojb *dstkey   为存储对比结果的 key
 * @param int  op        为运算操作类型
 */
void sunionDiffGenericCommand(client *c, robj **setkeys, 
        int setnum, robj *dstkey, int op) {
    // 申请集合空间
    robj **sets = zmalloc(sizeof(robj*)*setnum);
    // set 遍历迭代器
    setTypeIterator *si;
    robj *dstset = NULL;
    sds ele;
    int j, cardinality = 0;
    int diff_algo = 1;
    // 遍历每一个需要对比的集合，放入 集合数组中，方便后续运算使用
    for (j = 0; j < setnum; j++) {
        /**
         * 查找 key 的对象
         * lookupKeyRead 查找key，同时进行 过期检查 和 命中刷的统计
         * lookupKeyWrite 查找key，同时进行 过期检查
         */
        robj *setobj = dstkey ?
            lookupKeyWrite(c->db,setkeys[j]) :
            lookupKeyRead(c->db,setkeys[j]);
        // 如果集合不存在，则做 NULL 处理
        if (!setobj) {
            sets[j] = NULL;
            continue;
        }
        // 判断对象类型，如果对象不是集合则停止执行，并清理集合数组
        if (checkType(c,setobj,OBJ_SET)) {
            zfree(sets);
            return;
        }
        sets[j] = setobj;
    }
    
    /**
     * Redis 为 DIFF 操作提供了 两个算法
     * 算法1：复杂度为 O(n*m)，n 为第一个集合的元素个数，m 为 集合个数；逻辑为 对第一个集合进行遍历，对于它的每一个元素，依次在后面的所有集合中进行查找。只有在所有集合中都找不到的元素，才加入到最后的结果集合中
     * 算法2：复杂度为 O(n)，n 表示 所有集合元素之和；逻辑为 将第一个集合的所有元素都加入到一个中间集合中，遍历后面所有的集合，对于碰到的每一个元素，从中间集合中删掉它，最后中间集合剩下的元素就构成了差集
     * 这里会进行一些运算预估时间复杂度来决定 使用什么 算法
     */
    if (op == SET_OP_DIFF && sets[0]) {
        long long algo_one_work = 0, algo_two_work = 0;
        for (j = 0; j < setnum; j++) {
            if (sets[j] == NULL) continue;
            algo_one_work += setTypeSize(sets[0]);
            algo_two_work += setTypeSize(sets[j]);
        }
        // 算法1 在时间上和操作效率上具有更好的优势，在一定程度上优先选择第一种算法，因为它涉及到的操作比较少，只用添加，而第二种算法要先添加再删除；如果选择了第 1 种算法，那么在执行该算法之前，Redis 的实现中对于第二个集合之后的所有集合，按照元素个数由多到少进行了排序；这个排序有利于以更大的概率查找到元素，从而更快地结束查找
        algo_one_work /= 2;
        diff_algo = (algo_one_work <= algo_two_work) ? 1 : 2;
        if (diff_algo == 1 && setnum > 1) {
            qsort(sets+1,setnum-1,sizeof(robj*),
                qsortCompareSetsByRevCardinality);
        }
    }
    // 我们需要建立一个 临时的空集合
    dstset = createIntsetObject();
    // 并集
    if (op == SET_OP_UNION) {
        // 将每个元素添加至 tmp 集合中
        for (j = 0; j < setnum; j++) {
            // 跳过空集
            if (!sets[j]) continue;
            // 初始化 迭代器
            si = setTypeInitIterator(sets[j]);
            // 遍历集合，直接把元素加入到结果集
            while((ele = setTypeNextObject(si)) != NULL) {
                if (setTypeAdd(dstset,ele)) cardinality++;
                sdsfree(ele);
            }
            setTypeReleaseIterator(si);
        }
    } else if (op == SET_OP_DIFF && sets[0] && diff_algo == 1) {
    // 算法1 差集
        
        // 程序遍历集合set[0]中的所有元素，并将这个元素和其他集合中的所有元素进行对比，只有这个元素不存在于其他集合时，才会将它添加到结果集中
        si = setTypeInitIterator(sets[0]);
        while((ele = setTypeNextObject(si)) != NULL) {
            for (j = 1; j < setnum; j++) {
                // 跳过空集
                if (!sets[j]) continue;
                // 相同集合跳过
                if (sets[j] == sets[0]) break;
                // 对比元素是否在 j集合里
                if (setTypeIsMember(sets[j],ele)) break;
            }
            // 如果元素不存在其他任何集合中，添加到结果集
            if (j == setnum) {
                setTypeAdd(dstset,ele);
                cardinality++;
            }
            sdsfree(ele);
        }
        setTypeReleaseIterator(si);
    } else if (op == SET_OP_DIFF && sets[0] && diff_algo == 2) {
    // 算法2 差集
        
        // 如果是集合0，则添加到结果集合；如果非集合0，则将相同的元素从结果集合中移除，算法复杂度为O(N)，N为所有集合元素的总个
        for (j = 0; j < setnum; j++) {
            // 跳过空集
            if (!sets[j]) continue;
            si = setTypeInitIterator(sets[j]);
            // 遍历结果集
            while((ele = setTypeNextObject(si)) != NULL) {
                if (j == 0) {
                    // 将set[0]添加到结果集
                    if (setTypeAdd(dstset,ele)) cardinality++;
                } else {
                    // 将相同的元素从结果集中移除
                    if (setTypeRemove(dstset,ele)) cardinality--;
                }
                sdsfree(ele);
            }
            setTypeReleaseIterator(si);
            // 如果结果集是空的话就退出，因为set[0]中的元素都出现到其他集合中了
            if (cardinality == 0) break;
        }
    }
    // 输出结果集合
    if (!dstkey) {
    // 直接输出集合
    
        addReplyMultiBulkLen(c,cardinality);
        si = setTypeInitIterator(dstset);
        // 遍历并返回结果集合dstset中所有的元素
        while((ele = setTypeNextObject(si)) != NULL) {
            addReplyBulkCBuffer(c,ele,sdslen(ele));
            sdsfree(ele);
        }
        setTypeReleaseIterator(si);
        decrRefCount(dstset);
    } else {
    // 转储结果集
        
        // 删除原本库里面的 dstkey
        int deleted = dbDelete(c->db,dstkey);
        if (setTypeSize(dstset) > 0) {
            // 转储到目的 key
            dbAdd(c->db,dstkey,dstset);
            addReplyLongLong(c,setTypeSize(dstset));
            notifyKeyspaceEvent(NOTIFY_SET,
                op == SET_OP_UNION ? "sunionstore" : "sdiffstore",
                dstkey,c->db->id);
        } else {
            // 如果结果集合为空，删除
            decrRefCount(dstset);
            addReply(c,shared.czero);
            if (deleted)
                // 发送事件通知
                notifyKeyspaceEvent(NOTIFY_GENERIC,"del",
                    dstkey,c->db->id);
        }
        // 发送消息
        signalModifiedKey(c->db,dstkey);
        server.dirty++;
    }
    zfree(sets);
}

而另外还有一个交集操作，命令（sinter、sinterstore） 很值得探讨一下，这些命令的核心处理函数都为 sinterGenericCommand()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
/* t_set.c: line 794 */
// 交集标识
#define SET_OP_INTER 2
/**
 * @param robj **setkeys 为需要对比的 key 的集合
 * @param int  setnum    为需要对比的 key 的个数
 * @param rojb *dstkey   为存储对比结果的 key
 */
void sinterGenericCommand(client *c, robj **setkeys, unsigned long setnum, robj *dstkey) {
    // 申请集合空间
    robj **sets = zmalloc(sizeof(robj*)*setnum);
    // set 遍历迭代器
    setTypeIterator *si;
    robj *dstset = NULL;
    sds elesds;
    int64_t intobj;
    void *replylen = NULL;
    unsigned long j, cardinality = 0;
    int encoding;
    // 遍历每一个需要对比的集合，放入 集合数组中，方便后续运算使用
    for (j = 0; j < setnum; j++) {
        /**
         * 查找 key 的对象
         * lookupKeyRead 查找key，同时进行 过期检查 和 命中刷的统计
         * lookupKeyWrite 查找key，同时进行 过期检查
         */
        robj *setobj = dstkey ?
            lookupKeyWrite(c->db,setkeys[j]) :
            lookupKeyRead(c->db,setkeys[j]);
        // 如果集合不存在，则做 NULL 处理
        if (!setobj) {
            zfree(sets);
            // 如果有 目标 key
            if (dstkey) {
                // 删除目标 key 的内容
                if (dbDelete(c->db,dstkey)) {
                    signalModifiedKey(c->db,dstkey);
                    server.dirty++;
                }
                addReply(c,shared.czero);
            } else {
                addReply(c,shared.emptymultibulk);
            }
            return;
        }
        // 判断对象类型，如果对象不是集合则停止执行，并清理集合数组
        if (checkType(c,setobj,OBJ_SET)) {
            zfree(sets);
            return;
        }
        sets[j] = setobj;
    }
    // 从小到大排序集合数组中的集合大小，能够提高算法的性能
    qsort(sets,setnum,sizeof(robj*),qsortCompareSetsByCardinality);
    // 首先我们应该输出集合中元素的数量，但是现在不知道交集的大小，因此创建一个空对象的链表，然后保存所有的回复
    if (!dstkey) {
        // STINER命令创建一个链表
        replylen = addDeferredMultiBulkLength(c);
    } else {
        // STINERSTORE命令创建要给整数集合对
        dstset = createIntsetObject();
    }
    // 迭代第一个也是集合元素数量最小的集合的每一个元素，将该集合中的所有元素和其他集合作比较，如果至少有一个集合不包括该元素，则该元素不属于交集
    si = setTypeInitIterator(sets[0]);
    // 创建集合类型的迭代器并迭代器集合数组中的第一个集合的所有元素
    while((encoding = setTypeNext(si,&elesds,&intobj)) != -1) {
        // 遍历其他集合
        for (j = 1; j < setnum; j++) {
            // 跳过与第一个集合相等的集合，没有必要比较两个相同集合的元素，而且第一个集合作为结果的交集
            if (sets[j] == sets[0]) continue;
            // 当前元素为INTSET类型
            if (encoding == OBJ_ENCODING_INTSET) {
                // 如果在当前intset集合中没有找到该元素则直接跳过当前元素，操作下一个元素
                if (sets[j]->encoding == OBJ_ENCODING_INTSET &&
                    !intsetFind((intset*)sets[j]->ptr,intobj))
                {
                    break;
                } else if (sets[j]->encoding == OBJ_ENCODING_HT) {
                    // 在字典中查找
                    elesds = sdsfromlonglong(intobj);
                    // 如果当前元素不是当前集合中的元素，则释放字符串对象跳过for循环体，操作下一个元素
                    if (!setTypeIsMember(sets[j],elesds)) {
                        sdsfree(elesds);
                        break;
                    }
                    sdsfree(elesds);
                }
            } else if (encoding == OBJ_ENCODING_HT) {
            // 当前元素为HT字典类型
                
                // 在当前集合中查找该元素是否存在
                if (!setTypeIsMember(sets[j],elesds)) {
                    break;
                }
            }
        }
        // 执行到这里，该元素为结果集合中的元素
        if (j == setnum) {
            // 如果是SINTER命令，打印集合
            if (!dstkey) {
                if (encoding == OBJ_ENCODING_HT)
                    addReplyBulkCBuffer(c,elesds,sdslen(elesds));
                else
                    addReplyBulkLongLong(c,intobj);
                cardinality++;
            } else {
            // 如果是SINTERSTORE命令，先将结果添加到集合中，因为还要store 到数据库中
                if (encoding == OBJ_ENCODING_INTSET) {
                    elesds = sdsfromlonglong(intobj);
                    setTypeAdd(dstset,elesds);
                    sdsfree(elesds);
                } else {
                    setTypeAdd(dstset,elesds);
                }
            }
        }
    }
    setTypeReleaseIterator(si);
    // store 转储命令
    if (dstkey) {
    // 转储结果集
    
        // 删除原本库里面的 dstkey
        int deleted = dbDelete(c->db,dstkey);
        if (setTypeSize(dstset) > 0) {
            // 转储到目的 key
            dbAdd(c->db,dstkey,dstset);
            addReplyLongLong(c,setTypeSize(dstset));
            notifyKeyspaceEvent(NOTIFY_SET,"sinterstore",
                dstkey,c->db->id);
        } else {
            // 如果结果集合为空，删除
            decrRefCount(dstset);
            addReply(c,shared.czero);
            if (deleted)
                // 发送事件通知
                notifyKeyspaceEvent(NOTIFY_GENERIC,"del",
                    dstkey,c->db->id);
        }
        // 发送消息
        signalModifiedKey(c->db,dstkey);
        server.dirty++;
    } else {
        setDeferredMultiBulkLength(c,replylen,cardinality);
    }
    zfree(sets);
}

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-set.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

wettper <br /><br /> 现 离职状态 程序猿 <br /><br /> 叫醒我的不止阳光，还有梦想！<br /><br /> 喵喵、念念、健身、吉他！