【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中：

voidsetbitCommand(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偏移

intgetBitOffsetFromArgument(client *c, robj *o, size_t *offset, int hash, int bits) {
   longlong 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 # 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 # form. */
   if (usehash) loffset *= bits;

   /* 512MB字符串限制，不能超过 */
   if ((loffset < 0) || ((unsignedlonglong)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)) returnNULL;  //检测不是字符串类型返回
       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函数：

intswar(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命令

voidbitcountCommand(client *c) {
   robj *o;
   long start, end, strlen;
   unsignedchar *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;
   unsignedchar *p = s;
   uint32_t *p4;
   /**
   预先生成对应的bit表，因为一个字节非负数是可以是0～255个数字。所以生成了一个256数组。
   比如说0对应的二进制一个1都没有，bitsinbyte[0]则为0
   255对应二进制全是1，bitsinbyte[255]则为8
   */
   staticconstunsignedchar 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((unsignedlong)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 = (unsignedchar*)p4;
   while(count--) bits += bitsinbyte[*p++];
   return bits;
}



总结:

1)位图适合解决大数据存储减少存储资源。

2)setbit是通过sds字符串存储，是按照位存储。

3)bitcount采用SWAR算法，其实还采用了bitsinbyte预生成数组去减少计算开销。

4)setbit的最大限制是512MB