【Redis源码】位图SETBIT、BITCOUNT（九）

前言：

位图并不是一个特殊的数据结构，位图其实就是一个字符串，位图这种结构占用空间特别小。如果是数亿以上的用户如果存储活跃，如果是用key/value去存储每个节点都需要数G去存储，存储是很大问题。如果换做位图去存储则可以大大节约空间,不过适应于用户ID连续性的。除此之外也可以用作比如说点赞的存储。

（一）命令解析

setbit存储解析：

1）上图中操作setbit bitstr 0 1和setbit bitstr 7 1其实只占用1个字节，创建sds创建byte+1多加了一个字节。所以看到是2个字节。
2) 上图中操作setbit bitstr 25 1此时的byte等于4个字节，多加一个。所以是5个字节。
3) byte计算形式等于 offset / 8 。这个byte其实是计算存储到那个数组字节中。1个字节=8bit
4）计算存储在哪个bit里， (offset % 8 ) + 1。
如图：

（二）setbit源码解析

setbit命令，bitops.c中：

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)   //获得offset偏移,字符串有一个512MB的限制
        return;

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

    /* 当前值只能是1和0 */
    if (on & ~1) {
        addReplyError(c,err);
        return;
    }

    if ((o = lookupStringForBitCommand(c,bitoffset)) == NULL) return;   //查找或者创建字符串sds或扩容

    /* Get current values */
    byte = bitoffset >> 3;    //btye = bitoffset / 8 计算字节位置
    byteval = ((uint8_t*)o->ptr)[byte];  //获得存储到字节
    
    /*
    假设bitoffset=25。当前（1 << (7 - (bitoffset & 0x7))）= 64
    bit = 64对应二进制 = 0100 0000 
    bit用于做位运算
    */
    bit = 7 - (bitoffset & 0x7);          
    bitval = byteval & (1 << bit);       
    
    /* Update byte with new bit value and return original value */
    byteval &= ~(1 << bit);
    byteval |= ((on & 0x1) << bit);
    ((uint8_t*)o->ptr)[byte] = byteval;
    signalModifiedKey(c->db,c->argv[1]);
    notifyKeyspaceEvent(NOTIFY_STRING,"setbit",c->argv[1],c->db->id);
    server.dirty++;
    addReply(c, bitval ? shared.cone : shared.czero);
}

获得offset偏移

int getBitOffsetFromArgument(client *c, robj *o, size_t *offset, int hash, int bits) {
    long long loffset;
    char *err = "bit offset is not an integer or out of range";
    char *p = o->ptr;
    size_t plen = sdslen(p);
    int usehash = 0;

    /* Handle #<offset> form. */
    if (p[0] == '#' && hash && bits > 0) usehash = 1;

    if (string2ll(p+usehash,plen-usehash,&loffset) == 0) {
        addReplyError(c,err);
        return C_ERR;
    }

    /* Adjust the offset by 'bits' for #<offset> form. */
    if (usehash) loffset *= bits;

    /* 512MB字符串限制，不能超过 */
    if ((loffset < 0) || ((unsigned long long)loffset >> 3) >= (512*1024*1024))
    {
        addReplyError(c,err);
        return C_ERR;
    }

    *offset = (size_t)loffset;
    return C_OK;
}

查找或者创建字符串sds或扩容函数，bitops.c中：

robj *lookupStringForBitCommand(client *c, size_t maxbit) {
    size_t byte = maxbit >> 3;   //btye = bitoffset / 8 计算字节位置
    robj *o = lookupKeyWrite(c->db,c->argv[1]);

    if (o == NULL) {
        o = createObject(OBJ_STRING,sdsnewlen(NULL, byte+1));  //创建时多创建一个字节
        dbAdd(c->db,c->argv[1],o);
    } else {
        if (checkType(c,o,OBJ_STRING)) return NULL;  //检测不是字符串类型返回
        o = dbUnshareStringValue(c->db,c->argv[1],o); //获得robj对象
        o->ptr = sdsgrowzero(o->ptr,byte+1);  //内部使用了sdsMakeRoomFor重新计算sds字符串长度
    }
    return o;
}

（三）理解variable-precision SWAR算法

统计一个位数组中非0位的数量，数学上称作：”Hanmming Weight“(汉明重量)。最高的是variable-precision SWAR算法，可以在常数时间内计算出多个字节的非0数目。

swar函数：

int swar(uint32_t i)
{
    //计算每2位二进制数中1的个数
    i = ( i & 0x55555555) + ((i >> 1) & 0x55555555);
    //计算每4位二进制数中1的个数
    i = (i & 0x33333333) + ((i >> 2) & 0x33333333);
    //计算每8位二进制数中1的个数
    i = (i & 0x0F0F0F0F) + ((i >> 4) & 0x0F0F0F0F);
    //将每8位二进制数中1的个数和相加，并移至最低位8位
    i = (i * 0x01010101) >> 24;
    return i;
}

1）第一次计算

通过 ( i & 0x55555555) + ((i >> 1) & 0x55555555)计算后所得0x51618294，( i & 0x55555555) 得到每两位奇数的1的个数，而((i >> 1) & 0x55555555)在每两位上得到偶数的1个数。二者相加得到的就是每两位上1的个数。
每两位的取值范围二进制：01 、10、11。其实就是1、2、3。

2）第二次计算

通过 (i & 0x33333333) + ((i >> 2) & 0x33333333)计算后所得0x21312231。计算每4位的1的个数。

3）第三次计算

通过 (i & 0x0F0F0F0F) + ((i >> 2) & 0x0F0F0F0F)计算后所得0x3040404。计算每8位的1的个数。

4）第四次计算

0x3040404 * 0x0101010计算所得其实是一个long类型的数字0x3070b0f0c0804。但是由于uint32_t只占四位。多余部分会被移除所以只剩下0x0f0c0804，而0x0f0c0804对应的二进制是00001111 00001100 00001000 00000100 。右移动24位之后剩下00001111，而00001111对应的就是15。

（四）bitcount源码解析

bitcount命令

void bitcountCommand(client *c) {
    robj *o;
    long start, end, strlen;
    unsigned char *p;
    char llbuf[LONG_STR_SIZE];

    。。。省略
    
    /* Precondition: end >= 0 && end < strlen, so the only condition where
     * zero can be returned is: start > end. */
    if (start > end) {
        addReply(c,shared.czero);
    } else {
        long bytes = end-start+1;

        addReplyLongLong(c,redisPopcount(p+start,bytes)); //统计字节数组中的1数量
    }
}

统计字节数组中的1数量函数

size_t redisPopcount(void *s, long count) {
    size_t bits = 0;
    unsigned char *p = s;
    uint32_t *p4;
    /**
    预先生成对应的bit表，因为一个字节非负数是可以是0～255个数字。所以生成了一个256数组。
    比如说0对应的二进制一个1都没有，bitsinbyte[0]则为0
    255对应二进制全是1，bitsinbyte[255]则为8
    */
    static const unsigned char bitsinbyte[256] = {0,1,1,2,1,2,2,3,1,2,2,3,2,3,3,4,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,4,5,5,6,5,6,6,7,5,6,6,7,6,7,7,8};

    /* 计算对齐，以4字节对齐。多余部分先计算到bits中 */
    while((unsigned long)p & 3 && count) {
        bits += bitsinbyte[*p++];
        count--;  //减少统计字节数
    }

    /* 
    开始用variable-precision SWAR算法计算“1”的个数,
    计算以 28 字节宽度统计，节约计算时间
     */
    p4 = (uint32_t*)p;
    while(count>=28) { //统计字节 >= 28
        uint32_t aux1, aux2, aux3, aux4, aux5, aux6, aux7;

        aux1 = *p4++;
        aux2 = *p4++;
        aux3 = *p4++;
        aux4 = *p4++;
        aux5 = *p4++;
        aux6 = *p4++;
        aux7 = *p4++;
        count -= 28;

        aux1 = aux1 - ((aux1 >> 1) & 0x55555555);  //第一步
        aux1 = (aux1 & 0x33333333) + ((aux1 >> 2) & 0x33333333); //第二步
        aux2 = aux2 - ((aux2 >> 1) & 0x55555555);
        aux2 = (aux2 & 0x33333333) + ((aux2 >> 2) & 0x33333333);
        aux3 = aux3 - ((aux3 >> 1) & 0x55555555);
        aux3 = (aux3 & 0x33333333) + ((aux3 >> 2) & 0x33333333);
        aux4 = aux4 - ((aux4 >> 1) & 0x55555555);
        aux4 = (aux4 & 0x33333333) + ((aux4 >> 2) & 0x33333333);
        aux5 = aux5 - ((aux5 >> 1) & 0x55555555);
        aux5 = (aux5 & 0x33333333) + ((aux5 >> 2) & 0x33333333);
        aux6 = aux6 - ((aux6 >> 1) & 0x55555555);
        aux6 = (aux6 & 0x33333333) + ((aux6 >> 2) & 0x33333333);
        aux7 = aux7 - ((aux7 >> 1) & 0x55555555);
        aux7 = (aux7 & 0x33333333) + ((aux7 >> 2) & 0x33333333);
        bits += ((((aux1 + (aux1 >> 4)) & 0x0F0F0F0F) +
                    ((aux2 + (aux2 >> 4)) & 0x0F0F0F0F) +
                    ((aux3 + (aux3 >> 4)) & 0x0F0F0F0F) +
                    ((aux4 + (aux4 >> 4)) & 0x0F0F0F0F) +
                    ((aux5 + (aux5 >> 4)) & 0x0F0F0F0F) +
                    ((aux6 + (aux6 >> 4)) & 0x0F0F0F0F) +
                    ((aux7 + (aux7 >> 4)) & 0x0F0F0F0F))* 0x01010101) >> 24; //第三步
    }
    
    /* 计算剩余部分字节 */
    p = (unsigned char*)p4;
    while(count--) bits += bitsinbyte[*p++];
    return bits;
}

总结:

1)位图适合解决大数据存储减少存储资源。
2)setbit是通过sds字符串存储，是按照位存储。
3)bitcount采用SWAR算法，其实还采用了bitsinbyte预生成数组去减少计算开销。
4)setbit的最大限制是512MB

加V信进群交流