Redis源码阅读:bitmap 位图的运算

Redis提供了位图,位图也就是从bit的角度来看数据,可以把某一个bit设置为0或者1,举个简单的例子,我们希望记录用户某个月 的每一天是否登录,那么就只需要给一个32bit的值,如果第n天登录了,就把第n个bit设置为1。位图很节省内存,毕竟是从bit的 角度来看待和存储数据的,但是缺点也很明显,位图所需要存储的数据的大小取决于上限。

接下来我们看看Redis是怎么实现位图的,位运算比较难以理解,所以我们要用一个具体例子来进行辅助,我们将会阅读 SETBITGETBIT 两个命令来看源码。

    {"setbit",setbitCommand,4,
     "write use-memory @bitmap",
     0,NULL,1,1,1,0,0,0},

    {"getbit",getbitCommand,3,
     "read-only fast @bitmap",
     0,NULL,1,1,1,0,0,0},


/* SETBIT key offset bitvalue */
void setbitCommand(client *c) {
    robj *o;
    char *err = "bit is not an integer or out of range";
    size_t bitoffset;
    ssize_t byte, bit;
    int byteval, bitval;
    long on;

    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset,0,0) != C_OK)
        return;

    if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != C_OK)
        return;

    /* Bits can only be set or cleared... */
    if (on & ~1) {
        addReplyError(c,err);
        return;
    }

    if ((o = lookupStringForBitCommand(c,bitoffset)) == NULL) return;

    /* Get current values */
    // byte是指需要多少个字节来保存数据,向右位移3位,其实就是除以8,也就是从bit数得到byte数,因为一个byte等于8 bit
    // 当然,你看下面的运算的话,就可以看到,实际长度应该是 byte + 1,比如maxbit等于31,31 >> 3 = 3
    // 这实际上是保存对应bit 所在的byte的索引值,而长度应该是再加上1才对。下同。
    byte = bitoffset >> 3;
    // 得到对应byte的值,比如offset如果是31,那么byte就是3。如果第一次设置这个值,那么o->ptr应该是 00000000 00000000 00000000 00000001,byteval就是 00000001
    byteval = ((uint8_t*)o->ptr)[byte];

    // 得到对应bit的值,比如上面的运算,那么 应该是先计算括号里的值,
    // bitoffset & 0x7 就是 31 & 0x7 其实就是
    // 00011111 & 00000111 = 00000111 也就是0x7的值,
    // 然后计算 7 - 0x7,也就是 00000111 - 00000111 = 00000000
    // 这里其实就是计算出到底位于第几个bit
    // 我算了一下,如果bitoffset为30,那么值就是1,如果是29,那么值就是2:
    // >>> 7 - 31 & 0x7
    // 0
    // >>> 7 - 30 & 0x7
    // 1
    // >>> 7 - 29 & 0x7
    // 2

    bit = 7 - (bitoffset & 0x7);

    // 先计算 1 << 0,也就是1,二进制就是 00000001
    // 然后计算 byteval & 1,也就是 00000001 & 00000001 = 00000001
    // 得到只有那一个bit的时候的值
    bitval = byteval & (1 << bit);

    /* Update byte with new bit value and return original value */
    // 更新值,先找到对应的bit,然后取反,然后再和原值做&,
    // 按上面的例子, 1 << 0 也就是 00000001,取反以后是 11111110,byteval 是 00000001
    // 做 & 操作,得到 00000000
    // 这里其实就是把对应bit位上的值消掉,置为0,其它位不变。
    byteval &= ~(1 << bit);
    // 先执行 on & 0x1,如果我们是设置为1,那么这里就还是1,然后便宜bit位,得到 00000001
    // 然后 byteval 取或运算,也就是 00000000 | 00000001 = 00000001
    // 这里其实就是设置对应bit位的值
    byteval |= ((on & 0x1) << bit);
    // 然后把对应的byte设置成这个值,也就是 00000001,由于byte=3,也就是32bit中最后8个bit
    // 所以最后 o->ptr 所在值,其实就是 00000000 00000000 00000000 00000001
    ((uint8_t*)o->ptr)[byte] = byteval;
    signalModifiedKey(c,c->db,c->argv[1]);
    notifyKeyspaceEvent(NOTIFY_STRING,"setbit",c->argv[1],c->db->id);
    server.dirty++;
    addReply(c, bitval ? shared.cone : shared.czero);
}

/* GETBIT key offset */
void getbitCommand(client *c) {
    robj *o;
    char llbuf[32];
    size_t bitoffset;
    size_t byte, bit;
    size_t bitval = 0;

    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset,0,0) != C_OK)
        return;

    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,o,OBJ_STRING)) return;

    byte = bitoffset >> 3;
    bit = 7 - (bitoffset & 0x7);
    if (sdsEncodedObject(o)) {
        if (byte < sdslen(o->ptr))
            bitval = ((uint8_t*)o->ptr)[byte] & (1 << bit); // 取出对应值
    } else {
        if (byte < (size_t)ll2string(llbuf,sizeof(llbuf),(long)o->ptr))
            bitval = llbuf[byte] & (1 << bit);
    }

    addReply(c, bitval ? shared.cone : shared.czero);
}

/* This is an helper function for commands implementations that need to write
 * bits to a string object. The command creates or pad with zeroes the string
 * so that the 'maxbit' bit can be addressed. The object is finally
 * returned. Otherwise if the key holds a wrong type NULL is returned and
 * an error is sent to the client. */
robj *lookupStringForBitCommand(client *c, size_t maxbit) {
    size_t byte = maxbit >> 3;
    robj *o = lookupKeyWrite(c->db,c->argv[1]);
    if (checkType(c,o,OBJ_STRING)) return NULL;

    if (o == NULL) { // 如果没有这个对象,就创建并且保存
        o = createObject(OBJ_STRING,sdsnewlen(NULL, byte+1));
        dbAdd(c->db,c->argv[1],o);
    } else {
        // 如果有的话,看是否需要增加长度
        o = dbUnshareStringValue(c->db,c->argv[1],o);
        o->ptr = sdsgrowzero(o->ptr,byte+1);
    }
    return o;
}

总结

这一篇比较简单,我们仔细看了一下Redis是如何实现位图的,位运算比较难以理解, 我们以一个具体的例子进行了讲述。最后补充一句,除了最开始我们聚的例子,我们还可以用Redis的bitmap 来实现bloom filter,这里就不赘述了。


微信公众号
关注公众号,获得及时更新

更多文章
  • Go设计模式:模板模式
  • Go设计模式:适配器模式
  • Go设计模式:Iterator
  • glusterfs 笔记
  • 用peewee代替SQLAlchemy
  • Go的slice工作机制
  • Linux系统迁移记录(从HDD到SSD)
  • Redis是如何工作的?
  • virsh自动关闭windows虚拟机
  • Golang sort源码阅读
  • 分治的思维方式
  • Debian 使用NetworkManager之后networking.service崩溃
  • httprouter源码阅读与分析
  • 《程序员的自我修养-装载、链接与库》笔记
  • Golang sync.Pool源码阅读与分析