Redis之Set命令-白红宇

Redis之Set命令

阅读量：6925 次

发布时间：2019-06-27

本文共 12449 字，大约阅读时间需要 41 分钟。

0.前言

redis对无序集合的操作几个命令，本文介绍几个命令实际操作过程。

1.sadd命令

void saddCommand(redisClient *c) {    robj *set;    int j, added = 0;        /*查找集合，如果不存在创建新的集合*/    set = lookupKeyWrite(c->db,c->argv[1]);    if (set == NULL) {          /*          *创建集合，如果添加的元素可以转换为longlong类型，则存储格式采用intset数据结构，否则采用hash table数据结构进行存储          */        set = setTypeCreate(c->argv[2]);        dbAdd(c->db,c->argv[1],set);    } else {        if (set->type != REDIS_SET) {            addReply(c,shared.wrongtypeerr);            return;        }    }    for (j = 2; j < c->argc; j++) {        c->argv[j] = tryObjectEncoding(c->argv[j]);          /*元素添加进集合中*/        if (setTypeAdd(set,c->argv[j])) added++;    }    if (added) {        signalModifiedKey(c->db,c->argv[1]);        notifyKeyspaceEvent(REDIS_NOTIFY_SET,"sadd",c->argv[1],c->db->id);    }    server.dirty += added;    addReplyLongLong(c,added);}int setTypeAdd(robj *subject, robj *value) {    long long llval;    if (subject->encoding == REDIS_ENCODING_HT) {        if (dictAdd(subject->ptr,value,NULL) == DICT_OK) {            incrRefCount(value);            return 1;        }    } else if (subject->encoding == REDIS_ENCODING_INTSET) {          /*如果添加元素可以转换为longlong类型，保存至intset中，否则需要转换存储结构为hash table*/        if (isObjectRepresentableAsLongLong(value,&llval) == REDIS_OK) {            uint8_t success = 0;            subject->ptr = intsetAdd(subject->ptr,llval,&success);            if (success) {                /* 为了防止intset过大，set_max_intset_entries值作为一个阀值，占用空间大于此值，则将存储结构转换为hash table类型*/                if (intsetLen(subject->ptr) > server.set_max_intset_entries)                    setTypeConvert(subject,REDIS_ENCODING_HT);                return 1;            }        } else {            /* 转换为longlong失败，需要转换为hash table*/            setTypeConvert(subject,REDIS_ENCODING_HT);            /* 新元素添加至hash table中*/            redisAssertWithInfo(NULL,value,dictAdd(subject->ptr,value,NULL) == DICT_OK);            incrRefCount(value);            return 1;        }    } else {        redisPanic("Unknown set encoding");    }    return 0;}

2.求差集和并集命令(sdiff,sdiffstore,sunion,sunionstore)

sdiff求差集, sdiffstore求差集并保存结果, sunion求并集, sunionstore求并集并保存结果, 几种运算过程都是通过sunionDiffGenericCommand函数进行,此处将几个命令全部列出.

/*求并集*/void sunionCommand(redisClient *c) {    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_UNION);}/*求并集并保存结果*/void sunionstoreCommand(redisClient *c) {    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_UNION);}/*求差集*/void sdiffCommand(redisClient *c) {    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_DIFF);}/*求差集并保存结果*/void sdiffstoreCommand(redisClient *c) {    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_DIFF);}/*通用的求差集和并集函数*/void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *dstkey, int op) {    robj **sets = zmalloc(sizeof(robj*)*setnum);    setTypeIterator *si;    robj *ele, *dstset = NULL;    int j, cardinality = 0;    int diff_algo = 1;        /*取出需要操作的集合*/    for (j = 0; j < setnum; j++) {        robj *setobj = dstkey ?            lookupKeyWrite(c->db,setkeys[j]) :            lookupKeyRead(c->db,setkeys[j]);        if (!setobj) {            sets[j] = NULL;            continue;        }        if (checkType(c,setobj,REDIS_SET)) {            zfree(sets);            return;        }        sets[j] = setobj;    }    /*     *依据待运算集合中元素数量,选择计算差集算法, 其中算法1时间复杂度:O(N*M), N是第一个集合中元素个数, M是参与运算的集合数量.     *算法2时间复杂度:O(N), N是所有集合中元素数量总和     */    if (op == REDIS_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]);        }        /*          *algo_one_work值即为算法1中N*M, algo_two_work值即为算法2中N. 考虑到如果参与运算集合为intset时, 算法1的时间复杂度稳定性要好于算法2,          *因此没有直接比较两者大小选择算法, 而是算法1理论时间复杂度一半大于算法2时, 才使用算法2          */        algo_one_work /= 2;        diff_algo = (algo_one_work <= algo_two_work) ? 1 : 2;        if (diff_algo == 1 && setnum > 1) {            /*为了提高算法1速度, 尽快找到重复元素, 对集合列表按照元素数量进行了降序排序*/            qsort(sets+1,setnum-1,sizeof(robj*),                qsortCompareSetsByRevCardinality);        }    }    /*创建一个临时集合存放计算结果*/    dstset = createIntsetObject();    if (op == REDIS_OP_UNION) {        /* 求并集很简单了, 直接遍历所有元素, 添加进dstset集合中即可*/        for (j = 0; j < setnum; j++) {            if (!sets[j]) continue; /* non existing keys are like empty sets */            si = setTypeInitIterator(sets[j]);            while((ele = setTypeNextObject(si)) != NULL) {                if (setTypeAdd(dstset,ele)) cardinality++;                decrRefCount(ele);            }            setTypeReleaseIterator(si);        }    } else if (op == REDIS_OP_DIFF && sets[0] && diff_algo == 1) {        /*          *算法1对集合1进行遍历, 并判断集合1中的元素是否在其他集合中出现, 没有出现则添加到dstset集合中, 作为差集的一个元素          */        si = setTypeInitIterator(sets[0]);          /*          *循环外层对集合1进行遍历, 内层对其他参与运算的集合进行遍历          */        while((ele = setTypeNextObject(si)) != NULL) {            for (j = 1; j < setnum; j++) {                if (!sets[j]) continue; /* no key is an empty set. */                if (sets[j] == sets[0]) break; /* same set! */                if (setTypeIsMember(sets[j],ele)) break;            }            if (j == setnum) {                /* 其他集合中没有找到该元素, 添加到差集集合中*/                setTypeAdd(dstset,ele);                cardinality++;            }            decrRefCount(ele);        }        setTypeReleaseIterator(si);    } else if (op == REDIS_OP_DIFF && sets[0] && diff_algo == 2) {        /*          *算法2将集合1中元素直接copy进dstset集合中, 通过遍历其他所有集合, 然后确认其他集合中的元素没有在dstset中出现, 出现则从dstset中删除, 最终获取差集          */        for (j = 0; j < setnum; j++) {            if (!sets[j]) continue; /* non existing keys are like empty sets */            si = setTypeInitIterator(sets[j]);            while((ele = setTypeNextObject(si)) != NULL) {                if (j == 0) {                         /*集合1中元素添加进dstset中*/                    if (setTypeAdd(dstset,ele)) cardinality++;                } else {                         /*其他集合中元素出现在dstset中,则删除该元素*/                    if (setTypeRemove(dstset,ele)) cardinality--;                }                decrRefCount(ele);            }            setTypeReleaseIterator(si);            if (cardinality == 0) break;        }    }    if (!dstkey) {         /*运算结果不需要存储,直接返回结果元素至客户端*/        addReplyMultiBulkLen(c,cardinality);        si = setTypeInitIterator(dstset);        while((ele = setTypeNextObject(si)) != NULL) {            addReplyBulk(c,ele);            decrRefCount(ele);        }        setTypeReleaseIterator(si);        decrRefCount(dstset);    } else {        /* 需要存储, 首先删除原来可能已经存在dstkey的集合*/        int deleted = dbDelete(c->db,dstkey);        if (setTypeSize(dstset) > 0) {            dbAdd(c->db,dstkey,dstset);            addReplyLongLong(c,setTypeSize(dstset));            notifyKeyspaceEvent(REDIS_NOTIFY_SET,                op == REDIS_OP_UNION ? "sunionstore" : "sdiffstore",                dstkey,c->db->id);        } else {            decrRefCount(dstset);            addReply(c,shared.czero);            if (deleted)                notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del",                    dstkey,c->db->id);        }        signalModifiedKey(c->db,dstkey);        server.dirty++;    }    zfree(sets);}

3.求交集命令(sinter,sinterstore)

sinter求交集, sinterstore求交集并保存结果, 都是通过sinterGenericCommand函数进行相应的操作

/*求交集*/void sinterCommand(redisClient *c) {    sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);}/*求交集并保存结果*/void sinterstoreCommand(redisClient *c) {    sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);}/*通用求交集函数*/void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, robj *dstkey) {    robj **sets = zmalloc(sizeof(robj*)*setnum);    setTypeIterator *si;    robj *eleobj, *dstset = NULL;    int64_t intobj;    void *replylen = NULL;    unsigned long j, cardinality = 0;    int encoding;     /*遍历所有key, 读出所有传入的所有集合*/    for (j = 0; j < setnum; j++) {        robj *setobj = dstkey ?            lookupKeyWrite(c->db,setkeys[j]) :            lookupKeyRead(c->db,setkeys[j]);        if (!setobj) {            zfree(sets);            if (dstkey) {                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,REDIS_SET)) {            zfree(sets);            return;        }        sets[j] = setobj;    }    /* 按照集合中元素数量升序排列, 提高后面算法性能, 尽快决定元素是否是交集元素*/    qsort(sets,setnum,sizeof(robj*),qsortCompareSetsByCardinality);    /* The first thing we should output is the total number of elements...     * since this is a multi-bulk write, but at this stage we don't know     * the intersection set size, so we use a trick, append an empty object     * to the output list and save the pointer to later modify it with the     * right length */    if (!dstkey) {        replylen = addDeferredMultiBulkLength(c);    } else {        /* If we have a target key where to store the resulting set         * create this key with an empty set inside */        dstset = createIntsetObject();    }    /* Iterate all the elements of the first (smallest) set, and test     * the element against all the other sets, if at least one set does     * not include the element it is discarded */    si = setTypeInitIterator(sets[0]);    while((encoding = setTypeNext(si,&eleobj,&intobj)) != -1) {        for (j = 1; j < setnum; j++) {            if (sets[j] == sets[0]) continue;               /*               *依据不同的编码进行相应的操作               */            if (encoding == REDIS_ENCODING_INTSET) {                /* 编码均为intset时,则直接进行查找 */                if (sets[j]->encoding == REDIS_ENCODING_INTSET &&                    !intsetFind((intset*)sets[j]->ptr,intobj))                {                    break;                /* 编码为hash table时, 重新创建object进行比较 */                } else if (sets[j]->encoding == REDIS_ENCODING_HT) {                    eleobj = createStringObjectFromLongLong(intobj);                    if (!setTypeIsMember(sets[j],eleobj)) {                        decrRefCount(eleobj);                        break;                    }                    decrRefCount(eleobj);                }            } else if (encoding == REDIS_ENCODING_HT) {                    /*待查集合为intset, 则可以直接安卓long类型进行查找, 否则只能object在hash table中查找*/                if (eleobj->encoding == REDIS_ENCODING_INT &&                    sets[j]->encoding == REDIS_ENCODING_INTSET &&                    !intsetFind((intset*)sets[j]->ptr,(long)eleobj->ptr))                {                    break;                } else if (!setTypeIsMember(sets[j],eleobj)) {                    break;                }            }        }        /* 查找到最后一个集合表示此元素在所有集合中均出现, 作为交集结果 */        if (j == setnum) {            if (!dstkey) {                if (encoding == REDIS_ENCODING_HT)                    addReplyBulk(c,eleobj);                else                    addReplyBulkLongLong(c,intobj);                cardinality++;            } else {                if (encoding == REDIS_ENCODING_INTSET) {                    eleobj = createStringObjectFromLongLong(intobj);                    setTypeAdd(dstset,eleobj);                    decrRefCount(eleobj);                } else {                    setTypeAdd(dstset,eleobj);                }            }        }    }    setTypeReleaseIterator(si);         /*判断是否需要存储交集结果, 并进行相应操作*/    if (dstkey) {        int deleted = dbDelete(c->db,dstkey);        if (setTypeSize(dstset) > 0) {            dbAdd(c->db,dstkey,dstset);            addReplyLongLong(c,setTypeSize(dstset));            notifyKeyspaceEvent(REDIS_NOTIFY_SET,"sinterstore",                dstkey,c->db->id);        } else {            decrRefCount(dstset);            addReply(c,shared.czero);            if (deleted)                notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del",                    dstkey,c->db->id);        }        signalModifiedKey(c->db,dstkey);        server.dirty++;    } else {        setDeferredMultiBulkLength(c,replylen,cardinality);    }    zfree(sets);}